前言
很多文檔和博客都只介紹如何開發MCP Server，然後集成到VS Code或者Cursor等程序，很少涉及如何開發MCP Host和MCP Client。如果你想要在自己的服務中集成完整的MCP功能，光看這些是遠遠不夠的。所以本文及後續的MCP系列文章都會帶你深入瞭解如何開發MCP Client，讓你真正掌握這項技術。
準備開發環境
MCP官方SDK主要支持Python和TypeScript，當然也有其他語言的實現，不過我這裏就以Python為例了。我的Python版本是3.13.5，但其實只要高於3.11應該都沒問題。
我個人推薦使用uv來管理依賴，當然你也可以用傳統的pip。Python SDK有官方的mcp包和社區的FastMCP包。官方SDK其實也內置了FastMCP，不過是v1版本，而FastMCP官網已經更新到了v2版本。作為學習，兩個都裝上試試也無妨。
# 使用 uv
uv add mcp fastmcp
# 使用 pip
python -m pip install mcp fastmcp
第一個MCP項目：你好，MCP世界！
在第一個MCP項目中，我們實現一個簡單的MCP Client和MCP Server，但還沒集成LLM。在這個階段，Client調用Server的tool或resource都需要手動指定。
MCP Server
下面的MCP Server示例代碼定義了一些prompts、resources和tools。這裏有個小貼士：函數參數的類型註解、返回類型和docstring都一定要寫清楚，否則後續集成LLM時，LLM就無法正確理解如何調用你的工具了。
這段Server可以通過stdio方式被Client調用。在正式讓Client調用之前，建議你先手動運行一下Server，測試它能否正常啓動，避免Client啓動時報一堆讓人摸不着頭腦的錯誤。
from mcp.server.fastmcp import FastMCP
from datetime import datetime
import asyncssh
from typing import TypeAlias, Union
mcp = FastMCP("custom")
@mcp.prompt()
def greet_user(name: str, style: str = "formal") -> str:
"""Greet a user with a specified style."""
if style == "formal":
return f"Good day, {name}. How do you do?"
elif style == "friendly":
return f"Hey {name}! What's up?"
elif style == "casual":
return f"Yo {name}, how's it going?"
else:
return f"Hello, {name}!"
@mcp.resource("greeting://{name}")
def greeting_resource(name: str) -> str:
"""A simple greeting resource."""
return f"Hello, {name}!"
@mcp.resource("config://app")
def get_config() -> str:
"""Static configuration data"""
return "App configuration here"
@mcp.tool()
def add(a: int, b: int) -> int:
"""Add two numbers"""
return a + b
@mcp.tool()
def multiply(a: int, b: int) -> int:
"""Multiply two numbers"""
return a * b
Number: TypeAlias = Union[int, float]
@mcp.tool()
def is_greater_than(a: Number, b: Number) -> Number:
"""Check if a is greater than b"""
return a > b
@mcp.tool()
async def get_weather(city: str) -> str:
"""Get weather for a given city."""
return f"It's always sunny in {city}!"
@mcp.tool()
async def get_date() -> str:
"""Get today's date."""
return datetime.now().strftime("%Y-%m-%d")
@mcp.tool()
async def execute_ssh_command_remote(hostname: str, command: str) -> str:
"""Execute an SSH command on a remote host.
Args:
hostname (str): The hostname of the remote host.
command (str): The SSH command to execute.
Returns:
str: The output of the SSH command.
"""
async with asyncssh.connect(hostname, username="rainux", connect_timeout=10) as conn:
result = await conn.run(command, timeout=10)
stdout = result.stdout
stderr = result.stderr
content = str(stdout if stdout else stderr)
return content
if __name__ == "__main__":
mcp.run(transport="stdio")
MCP Client
Client通過STDIO方式調用MCP Server，server_params中指定了如何運行Server，包括python解釋器路徑、Server文件名和運行位置。需要注意的是，Client啓動時也會啓動Server，如果Server報錯，Client也會跟着無法啓動。
import asyncio
from pathlib import Path
from pydantic import AnyUrl
from mcp import ClientSession, StdioServerParameters, types
from mcp.client.stdio import stdio_client
server_params = StdioServerParameters(
command=str(Path(__file__).parent / ".venv" / "bin" / "python"),
args=[str(Path(__file__).parent / "demo1-server.py")],
cwd=str(Path(__file__).parent),
)
async def run():
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
# Initialize the connection
await session.initialize()
# List available prompts
prompts = await session.list_prompts()
print(f"Available prompts: {[p.name for p in prompts.prompts]}")
# Get a prompt (greet_user prompt from fastmcp_quickstart)
if prompts.prompts:
prompt = await session.get_prompt("greet_user", arguments={"name": "Alice", "style": "friendly"})
print(f"Prompt result: {prompt.messages[0].content}")
# List available resources
resources = await session.list_resources()
print(f"Available resources: {[r.uri for r in resources.resources]}")
# List available tools
tools = await session.list_tools()
print(f"Available tools: {[t.name for t in tools.tools]}")
# Read a resource (greeting resource from fastmcp_quickstart)
resource_content = await session.read_resource(AnyUrl("greeting://World"))
content_block = resource_content.contents[0]
if isinstance(content_block, types.TextResourceContents):
print(f"Resource content: {content_block.text}")
# Call a tool (add tool from fastmcp_quickstart)
result = await session.call_tool("add", arguments={"a": 5, "b": 3})
result_unstructured = result.content[0]
if isinstance(result_unstructured, types.TextContent):
print(f"Tool result: {result_unstructured.text}")
result_structured = result.structuredContent
print(f"Structured tool result: {result_structured}")
if __name__ == "__main__":
asyncio.run(run())
運行Client，輸出如下：
Processing request of type ListPromptsRequest
Available prompts: ['greet_user']
Processing request of type GetPromptRequest
Prompt result: type='text' text="Hey Alice! What's up?" annotations=None meta=None
Processing request of type ListResourcesRequest
Available resources: [AnyUrl('config://app')]
Processing request of type ListToolsRequest
Available tools: ['add', 'multiply', 'get_weather', 'get_date', 'execute_ssh_command_remote']
Processing request of type ReadResourceRequest
Resource content: Hello, World!
Processing request of type CallToolRequest
Tool result: 8
Structured tool result: {'result': 8}
可以看到，Client成功地調用了Server上的各種功能，包括獲取提示、讀取資源和調用工具。
使用streamable-http遠程調用：讓MCP飛起來！
上面的例子中，Client通過STDIO方式在本地調用Server。現在我們稍作修改，讓它可以通過HTTP遠程調用Server，這樣就更加靈活了。
MCP Server
只列出修改的部分：
mcp = FastMCP("custom", host="localhost", port=8001)
if __name__ == "__main__":
mcp.run(transport="streamable-http")
修改完成後，啓動Server，它會監聽在localhost:8001地址上，就像一個小小的Web服務（其實就是個Web服務，暴露的api為/mcp）。
MCP Client
同樣只列出修改的部分。Client需要指定MCP Server的地址。streamablehttp_client返回的第三個參數get_session_id用於會話管理，大多數情況下你不需要直接使用它，所以在一些文檔中這裏會用_來佔位。
from mcp.client.streamable_http import streamablehttp_client
server_uri = "http://localhost:8001/mcp"
async def main():
async with streamablehttp_client(server_uri) as (read, write, get_session_id):
# 獲取當前會話ID
session_id = get_session_id()
print(f"Session ID before initialization: {session_id}")
async with ClientSession(read, write) as session:
# Initialize the connection
await session.initialize()
# 初始化後再次獲取會話ID
session_id = get_session_id()
print(f"Session ID after initialization: {session_id}")
client運行輸出：
Session ID before initialization: None
Session ID after initialization: 60ce4204b907469e9eb46e7e01df040d
Available prompts: ['greet_user']
Prompt result: type='text' text="Hey Alice! What's up?" annotations=None meta=None
Available resources: [AnyUrl('config://app')]
Available tools: ['add', 'multiply', 'get_weather', 'get_date', 'execute_ssh_command_remote']
Resource content: Hello, World!
Tool result: 8
Structured tool result: {'result': 8}
現在我們的MCP應用已經可以通過網絡進行遠程調用了，架構變得更加靈活。
集成LLM：讓AI自己做決定！
前面兩個示例中，我們都需要在Client中手動控制調用Server的tool，這在實際應用中顯然是不現實的。我們需要集成LLM，讓AI自己決定該調用哪個工具。
MCP Server
Server端不需要做任何變更，Client還是通過HTTP方式調用我們之前創建的Server。
MCP Client
這裏我們選用阿里的通義千問（Qwen）。Qwen的API Key可以自行申請，氪個5塊錢就夠個人開發用很久了。為了便於後續開發，我把配置功能單獨放到了一個模塊裏，下面代碼中直接使用了，相關模塊放在"補充"部分。
"""
MCP (Model Context Protocol) 客户端示例
該客户端演示瞭如何使用 MCP 協議與 MCP 服務器進行交互，並通過 LLM 調用服務器提供的工具。
工作流程：
1. 連接到 MCP 服務器
2. 獲取服務器提供的工具列表
3. 用户輸入查詢
4. 將查詢發送給 LLM，LLM 可能會調用 MCP 服務器提供的工具
5. 執行工具調用並獲取結果
6. 將結果返回給 LLM 進行最終回答
"""
import asyncio
# JSON 處理
import json
# 增強輸入功能（在某些系統上提供命令歷史等功能）
import readline # 引入readline模塊用於增強python的input功能, Windows下的python標準庫可能不包含
# 異常追蹤信息
import traceback
# 異步上下文管理器，用於資源管理
from contextlib import AsyncExitStack
# 類型提示支持
from typing import List, Optional, cast
# MCP 客户端會話和 HTTP 傳輸
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
# OpenAI 異步客户端，用於與 LLM 通信
from openai import AsyncOpenAI
# OpenAI 聊天完成相關的類型定義
from openai.types.chat import (ChatCompletionAssistantMessageParam,
ChatCompletionMessageFunctionToolCall,
ChatCompletionMessageParam,
ChatCompletionMessageToolCall,
ChatCompletionToolMessageParam,
ChatCompletionToolParam,
ChatCompletionUserMessageParam)
# 項目配置和日誌模塊
from pkg.config import cfg
from pkg.log import logger
class MCPClient:
"""
MCP 客户端類，負責管理與 MCP 服務器的連接和交互
"""
def __init__(self):
"""
初始化 MCP 客户端
"""
# 客户端會話，初始為空
self.session: Optional[ClientSession] = None
# 異步上下文管理棧，用於管理異步資源的生命週期
self.exit_stack = AsyncExitStack()
# OpenAI 異步客户端，用於與 LLM 通信
self.client = AsyncOpenAI(
base_url=cfg.llm_base_url,
api_key=cfg.llm_api_key,
)
async def connect_to_server(self, server_uri: str):
"""
連接到 MCP 服務器
Args:
server_uri (str): MCP 服務器的 URI
"""
# 創建 Streamable HTTP 傳輸連接
http_transport = await self.exit_stack.enter_async_context(
streamablehttp_client(server_uri)
)
# 獲取讀寫流
self.read, self.write, _ = http_transport
# 創建並初始化客户端會話
self.session = await self.exit_stack.enter_async_context(
ClientSession(self.read, self.write)
)
# 初始化會話
await self.session.initialize()
# 檢查會話是否成功初始化
if self.session is None:
raise RuntimeError("Failed to initialize session")
# 獲取服務器提供的工具列表
response = await self.session.list_tools()
tools = response.tools
logger.info(f"\nConnected to server with tools: {[tool.name for tool in tools]}")
async def process_query(self, query: str) -> str:
"""
處理用户查詢
Args:
query (str): 用户的查詢
Returns:
str: 處理結果
"""
# 初始化消息歷史，包含用户的查詢
messages: List[ChatCompletionMessageParam] = [
ChatCompletionUserMessageParam(
role="user",
content=query
)
]
# 確保會話已初始化
if self.session is None:
raise RuntimeError("Session not initialized. Please connect to server first.")
# 獲取服務器提供的工具列表
response = await self.session.list_tools()
# 構建工具列表，處理可能為None的字段
# 這些工具將被傳遞給 LLM，以便 LLM 知道可以調用哪些工具
available_tools: List[ChatCompletionToolParam] = []
for tool in response.tools:
tool_def: ChatCompletionToolParam = {
"type": "function",
"function": {
"name": tool.name,
"description": tool.description or "",
"parameters": tool.inputSchema or {}
}
}
available_tools.append(tool_def)
logger.info(f"Available tools: {available_tools}")
# 調用 LLM 進行聊天完成
response = await self.client.chat.completions.create(
model=cfg.llm_model,
messages=messages,
tools=available_tools,
)
# 存儲最終輸出文本
final_text = []
# 獲取 LLM 的響應消息
message = response.choices[0].message
final_text.append(message.content or "")
# 如果 LLM 要求調用工具，則處理工具調用
while message.tool_calls:
# 處理每個工具調用
for tool_call in message.tool_calls:
# 確保我們處理的是正確的工具調用類型
if hasattr(tool_call, 'function'):
# 這是一個函數工具調用
function_call = cast(ChatCompletionMessageFunctionToolCall, tool_call)
function = function_call.function
tool_name = function.name
# 解析工具參數
tool_args = json.loads(function.arguments)
else:
# 跳過不支持的工具調用類型
continue
# 執行工具調用
if self.session is None:
raise RuntimeError("Session not initialized. Cannot call tool.")
# 調用 MCP 服務器上的工具
result = await self.session.call_tool(tool_name, tool_args)
final_text.append(f"[Calling tool {tool_name} with args {tool_args}]")
# 將工具調用和結果添加到消息歷史
# 這樣 LLM 可以知道它之前調用了哪些工具
assistant_msg: ChatCompletionAssistantMessageParam = {
"role": "assistant",
"tool_calls": [
{
"id": tool_call.id,
"type": "function",
"function": {
"name": tool_name,
"arguments": json.dumps(tool_args)
}
}
]
}
messages.append(assistant_msg)
# 添加工具調用結果到消息歷史
tool_msg: ChatCompletionToolMessageParam = {
"role": "tool",
"tool_call_id": tool_call.id,
"content": str(result.content) if result.content else ""
}
messages.append(tool_msg)
# 將工具調用的結果交給 LLM，讓 LLM 生成最終回答
response = await self.client.chat.completions.create(
model=cfg.llm_model,
messages=messages,
tools=available_tools
)
# 獲取新的響應消息
message = response.choices[0].message
if message.content:
final_text.append(message.content)
# 返回最終結果
return "\n".join(final_text)
async def chat_loop(self):
"""
運行交互式聊天循環
"""
print("\nMCP Client Started!")
print("Type your queries or 'quit' to exit.")
# 持續接收用户輸入
while True:
try:
# 獲取用户輸入
query = input("\nQuery: ").strip()
# 檢查是否退出
if query.lower() == 'quit':
break
# 忽略空輸入
if not query:
continue
# 處理用户查詢並輸出結果
response = await self.process_query(query)
print("\n" + response)
# 異常處理
except Exception as e:
print(f"\nError: {str(e)}")
print(traceback.format_exc())
async def cleanup(self):
"""
清理資源
"""
await self.exit_stack.aclose()
async def main():
"""
主函數
"""
# 創建 MCP 客户端實例
client = MCPClient()
try:
# 連接到 MCP 服務器
await client.connect_to_server("http://localhost:8001/mcp")
# 運行聊天循環
await client.chat_loop()
except Exception as e:
print(f"Error: {str(e)}")
finally:
# 清理資源
await client.cleanup()
# 程序入口點
if __name__ == "__main__":
asyncio.run(main())
client運行輸出：
MCP Client Started!
Type your queries or 'quit' to exit.
Query: 今天的日期是什麼
[Calling tool get_date with args {}]
今天的日期是2025年9月13日。
Query: 合肥的天氣怎麼樣？
[Calling tool get_weather with args {'city': '合肥'}]
合肥的天氣總是陽光明媚！
Query: 0.11比0.9大嗎
[Calling tool is_greater_than with args {'a': 0.11, 'b': 0.9}]
0.11 不比 0.9 大。0.11 小於 0.9。
Query: quit
現在AI可以自己決定調用哪個工具了。當你問"今天的日期是什麼"時，它會自動調用get_date工具；當你問"合肥的天氣怎麼樣"時，它會自動調用get_weather工具。這才是真正的智能！
小結
通過這篇文章，我們從零開始構建了一個完整的MCP應用，涵蓋了從基礎的Client-Server通信到集成LLM的全過程。我們學習了：
- 如何搭建MCP開發環境
- 如何創建MCP Server並定義tools、resources和prompts
- 如何編寫MCP Client並通過stdio和HTTP兩種方式與Server通信
- 如何集成LLM，讓AI自主決定調用哪個工具
整個過程就像搭積木一樣，每一步都有其特定的作用：
- Server負責提供功能（工具和資源）
- Client負責協調和調用這些功能
- LLM負責智能決策，決定何時以及如何使用這些功能
這種架構的優勢在於功能擴展非常靈活。當你需要添加新功能時，只需要在Server端添加新的tools或resources，Client和LLM會自動發現並使用它們，而不需要修改Client端的代碼。
MCP真正實現了"上下文協議"的概念，讓AI可以像人類一樣訪問和操作各種工具和資源，這是邁向更強大AI應用的重要一步。接下來你可以嘗試添加更多有趣的工具，比如文件操作、數據庫查詢、API調用等，讓你的AI助手變得更加強大！
補充
配置模塊
pkg/config.py
import json
from pathlib import Path
class Config:
def __init__(self):
p = Path(__file__).parent.parent / "conf" / "config.json"
if not p.exists():
raise FileNotFoundError(f"Config file not found: {p}")
self.data = self.read_json(str(p))
def read_json(self, filepath: str) -> dict:
with open(filepath, "r") as f:
return json.load(f)
@property
def llm_model(self) -> str:
return self.data["llm"]["model"]
@property
def llm_api_key(self):
return self.data["llm"]["api_key"]
@property
def llm_base_url(self) -> str:
return self.data["llm"]["base_url"]
@property
def server_host(self) -> str:
return self.data["server"]["host"]
@property
def server_port(self) -> int:
return self.data["server"]["port"]
cfg = Config()
配置文件conf/config.json
{
"llm": {
"model": "qwen-plus",
"base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1",
"api_key": "your token"
},
"server": {
"host": "127.0.0.1",
"port": 8000
}
}
日誌模塊
pkg/log.py
import logging
import sys
def set_formatter():
"""設置formatter"""
fmt = "%(asctime)s | %(name)s | %(levelname)s | %(filename)s:%(lineno)d | %(funcName)s | %(message)s"
datefmt = "%Y-%m-%d %H:%M:%S"
return logging.Formatter(fmt, datefmt=datefmt)
def set_stream_handler():
return logging.StreamHandler(sys.stdout)
def set_file_handler():
return logging.FileHandler("app.log", mode="a", encoding="utf-8")
def get_logger(name: str = "mylogger", level=logging.DEBUG):
logger = logging.getLogger(name)
formatter = set_formatter()
# handler = set_stream_handler()
handler = set_file_handler()
handler.setFormatter(formatter)
logger.addHandler(handler)
logger.setLevel(level)
return logger
logger = get_logger()
