在 MCP Server 开发实战：无缝对接 LLM 和 Elasticsearch 一文中，我们详细介绍了如何利用 MCP Python SDK 编写一个 Elasticsearch MCP 服务器，并通过 Claude Desktop 作为 MCP 客户端进行交互。本文将进一步介绍如何使用 MCP Python SDK 编写一个 MCP 客户端，以便更加灵活地与 MCP 服务器进行通信和集成。本文的完整代码可以在 Github 上找到：https://github.com/cr7258/hands-on-lab/tree/main/ai/claude/mcp/client/elasticsearch-mcp-client-example
MCP 系列文章：
一文带你入门 MCP（模型上下文协议） MCP Server 开发实战：无缝对接 LLM 和 Elasticsearch 快速上手：实现你的第一个 MCP Client（本文）
MCP 客户端的作用
MCP 客户端充当 LLM 和 MCP 服务器之间的桥梁，MCP 客户端的工作流程如下：
MCP 客户端首先从 MCP 服务器获取可用的工具列表。 将用户的查询连同工具描述通过 function calling 一起发送给 LLM。 LLM 决定是否需要使用工具以及使用哪些工具。 如果需要使用工具，MCP 客户端会通过 MCP 服务器执行相应的工具调用。 工具调用的结果会被发送回 LLM。 LLM 基于所有信息生成自然语言响应。 最后将响应展示给用户。
MCP 通信方式
MCP 支持两种通信方式：
标准输入输出（Standard Input/Output, stdio）：客户端通过启动服务器子进程并使用标准输入（stdin）和标准输出（stdout）建立双向通信，一个服务器进程只能与启动它的客户端通信（1:1 关系）。stdio 适用于本地快速集成的场景，在本文中，我们将使用这种方式来编写 MCP 客户端。
服务器发送事件（Server-Sent Events, SSE）：服务器作为独立进程运行，客户端和服务器代码完全解耦，支持多个客户端随时连接和断开。这种方式将在后续的系列文章中单独进行介绍。
简单的示例
在开始构建复杂的应用之前，让我们先创建一个最简单的 MCP 客户端。这个基础示例将展示如何连接到 MCP 服务器并使用其提供的工具。
MCP 服务器的代码以及 Elasticsearch 集群的配置文件请参考 MCP Server 开发实战：无缝对接 LLM 和 Elasticsearch 一文。
添加依赖
在本教程中，我们将使用 MCP Python SDK 来编写项目，使用 uv 来管理 Python 项目依赖。需要添加如下依赖：
uv add mcp elasticsearch openai
设置服务器连接参数
在使用 stdio
方式进行通信时，MCP 服务器的进程由 MCP 客户端程序负责启动。因此，我们通过 StdioServerParameters
来配置服务器进程的启动参数，包括运行 MCP 服务器的命令及其对应的参数。其中，sys.argv[1]
代表客户端程序运行时传入的第一个参数，用于指定服务器脚本的路径，从而确保 MCP 客户端能够正确启动并连接到 MCP 服务器。
server_params = StdioServerParameters(
command="python", # 运行命令
args=[sys.argv[1]], # 服务器脚本路径
env=None # 可选的环境变量
)
建立服务器连接
stdio_client
负责启动服务器进程并建立双向通信通道，它返回用于读写数据的流对象。ClientSession
则在这些流的基础上提供高层的会话管理，包括初始化连接、维护会话状态等。
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
调用工具
在 MCP 客户端中，我们使用了以下两个函数来与 MCP 服务器进行交互。
list_tools()
：获取 MCP 服务器提供的所有可用工具。call_tool(name, args)
：调用指定的工具并获取结果，这里调用list_indices
来获取 Elasticsearch 集群中的索引信息。
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
asyncdef run():
# 建立连接
asyncwith stdio_client(server_params) as (read, write):
asyncwith ClientSession(read, write) as session:
# 初始化连接
await session.initialize()
# 列出可用工具
tools = await session.list_tools()
print("Tools:", tools)
# 调用工具
indices = await session.call_tool("list_indices")
print("Indices:", indices)
if __name__ == "__main__":
import asyncio
asyncio.run(run())
运行程序
使用以下命令运行 MCP 客户端：
uv run simple.py <MCP 服务器的代码路径>
程序首先会列出所有可用的工具，可以看到总共有 3 个工具：list_indices
、get_index
和 write_documents
。
Tools:
meta=None
nextCursor=None
tools=[
Tool(
name='list_indices',
description='List all Elasticsearch indices',
inputSchema={
'properties': {},
'title': 'list_indicesArguments',
'type': 'object'
}
),
Tool(
name='get_index',
description='Get detailed information about a specific Elasticsearch index',
inputSchema={
'properties': {
'index': {
'title': 'Index',
'type': 'string'
}
},
'required': ['index'],
'title': 'get_indexArguments',
'type': 'object'
}
),
Tool(
name='write_documents',
description='Write multiple documents to an Elasticsearch index using bulk API',
inputSchema={
'properties': {
'index': {
'title': 'Index',
'type': 'string'
},
'documents': {
'items': {
'type': 'object'
},
'title': 'Documents',
'type': 'array'
}
},
'required': ['index', 'documents'],
'title': 'write_documentsArguments',
'type': 'object'
}
)
]
然后，程序会调用 list_indices
工具，获取 Elasticsearch 集群中的索引信息，并打印出来。
Indices:
meta=None
content=[
TextContent(
type='text',
text="[
{'health': 'green', 'status': 'open', 'index': 'student', 'uuid': 'iKl3j9ujSJ2i0GF5v8NUDw',
'pri': '1', 'rep': '1', 'docs.count': '3', 'docs.deleted': '0', 'store.size': '12kb', 'pri.store.size': '6kb'},
{'health': 'green', 'status': 'open', 'index': 'teacher', 'uuid': 'vfbwvH7yQGWqRGEG-t4FnA',
'pri': '1', 'rep': '1', 'docs.count': '3', 'docs.deleted': '0', 'store.size': '22.6kb', 'pri.store.size': '11.3kb'},
{'health': 'green', 'status': 'open', 'index': 'movies', 'uuid': 'JYhacHmXREWmkvwXBaFPmg',
'pri': '1', 'rep': '1', 'docs.count': '3', 'docs.deleted': '0', 'store.size': '13.2kb', 'pri.store.size': '6.6kb'}
]"
)
]
isError=False
进阶示例：集成 LLM
在实际应用中，我们通常希望让 LLM（如 OpenAI、Claude、通义千问等）自主决定调用哪些工具。下面的代码将以通义千问为示例进行演示，并使用 OpenAI SDK 与其交互。为了简化这一过程，我们将借助 OpenRouter -- 一个统一的 LLM 网关，它提供了 OpenAI 兼容的接口，使我们能够通过相同的 OpenAI API 访问包括通义千问在内的多种 LLM。
OpenRouter 的使用方式非常简单。我们只需在创建 OpenAI 客户端时指定 OpenRouter 的 base_url
和 api_key
，并在调用模型时以 <provider>/<model>
的格式（例如 qwen/qwen-plus
）指定目标模型，OpenRouter 就会根据模型名称自动将请求路由到对应的 LLM 上。除此之外，其他代码与标准的 OpenAI SDK 保持一致。关于 OpenRouter 的使用方法可以参考这里。
接下来介绍一下 MCP 客户端的主要代码。
初始化客户端类
MCPClient
类的初始化包含以下 3 个组件：
self.session
：用于存储与 MCP 服务器的会话对象，初始设为None
，将在连接服务器时被赋值。self.exit_stack：使用 AsyncExitStack
来管理异步资源，确保所有资源（如服务器连接、会话等）在程序结束时能够正确关闭。self.client：创建 OpenAI 异步客户端，通过 OpenRouter 来访问 LLM。这里我们：
设置 base_url
为 OpenRouter 的 API 端点。从环境变量获取 API Key（请确保设置了 OPENROUTER_API_KEY
环境变量）。可以参考该文档创建 OpenRouter 的 API Key。
class MCPClient:
def __init__(self):
self.session: Optional[ClientSession] = None
self.exit_stack = AsyncExitStack()
self.client = AsyncOpenAI(
base_url="https://openrouter.ai/api/v1",
api_key=os.getenv("OPENROUTER_API_KEY"),
)
服务器连接
connect_to_server
方法负责建立与 MCP 服务器的连接。它首先配置服务器进程的启动参数，然后通过 stdio_client
建立双向通信通道，最后创建并初始化会话。所有的资源管理都通过 AsyncExitStack
来处理，确保资源能够正确释放。连接成功后，它会打印出 MCP 服务器提供的所有可用工具。
async def connect_to_server(self, server_script_path: str):
server_params = StdioServerParameters(
command="python",
args=[server_script_path],
env=None
)
stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
self.stdio, self.write = stdio_transport
self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))
await self.session.initialize()
# 列出可用工具
response = await self.session.list_tools()
tools = response.tools
print("\nConnected to server with tools:", [tool.name for tool in tools])
处理请求
process_query
方法中定义了处理请求的流程：
首先，将用户的查询作为初始消息发送给 LLM，同时提供 MCP 服务器上所有可用工具的描述信息。 LLM 分析用户查询，决定是直接回答还是需要调用工具。如果需要工具，它会指定要调用的工具名称和参数。 对于每个工具调用，MCP 客户端执行调用并收集结果。 将工具调用的结果返回给 LLM，让它基于这些新信息生成或更新回答。 如果 LLM 认为还需要更多信息，它会继续请求调用其他工具。这个过程会一直重复，直到 LLM 收集了足够的信息来完整回答用户的查询。
async def process_query(self, query: str) -> str:
"""使用 LLM 和 MCP 服务器提供的工具处理查询"""
messages = [
{
"role": "user",
"content": query
}
]
response = await self.session.list_tools()
available_tools = [{
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": tool.inputSchema
}
} for tool in response.tools]
# 初始化 LLM API 调用
response = await self.client.chat.completions.create(
model="qwen/qwen-plus",
messages=messages,
tools=available_tools
)
final_text = []
message = response.choices[0].message
final_text.append(message.content or"")
# 处理响应并处理工具调用
while message.tool_calls:
# 处理每个工具调用
for tool_call in message.tool_calls:
tool_name = tool_call.function.name
tool_args = json.loads(tool_call.function.arguments)
# 执行工具调用
result = await self.session.call_tool(tool_name, tool_args)
final_text.append(f"[Calling tool {tool_name} with args {tool_args}]")
# 将工具调用和结果添加到消息历史
messages.append({
"role": "assistant",
"tool_calls": [
{
"id": tool_call.id,
"type": "function",
"function": {
"name": tool_name,
"arguments": json.dumps(tool_args)
}
}
]
})
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": str(result.content)
})
# 将工具调用的结果交给 LLM
response = await self.client.chat.completions.create(
model="qwen/qwen-plus",
messages=messages,
tools=available_tools
)
message = response.choices[0].message
if message.content:
final_text.append(message.content)
return"\n".join(final_text)
交互式界面
chat_loop
方法提供了一个简单的命令行交互界面：
在一个循环中持续接收用户输入，并通过前面定义地 process_query
方法处理查询。将处理结果格式化后显示给用户。 通过异常处理机制优雅地处理可能出现的错误。 用户输入 quit
时安全退出程序。
async def chat_loop(self):
"""运行交互式聊天循环"""
print("\nMCP Client Started!")
print("Type your queries or 'quit' to exit.")
whileTrue:
try:
query = input("\nQuery: ").strip()
if query.lower() == 'quit':
break
response = await self.process_query(query)
print("\n" + response)
except Exception as e:
print(f"\nError: {str(e)}")
程序入口
程序入口代码完成以下工作：
检查命令行参数，确保用户提供了服务器脚本的路径。 创建 MCPClient
实例并连接到服务器。启动交互式聊天循环。
使用 try-finally
结构确保在程序退出时正确清理资源。通过 asyncio.run()
运行异步主函数。
async def main():
if len(sys.argv) < 2:
print("Usage: uv run client.py <path_to_server_script>")
sys.exit(1)
client = MCPClient()
try:
await client.connect_to_server(sys.argv[1])
await client.chat_loop()
finally:
await client.cleanup()
if __name__ == "__main__":
asyncio.run(main())
运行程序
使用以下命令运行 MCP 客户端：
uv run client.py <MCP 服务器的代码路径>
启动 MCP 客户端后，我们输入问题 Elasticsearch 集群中有哪些索引?
，可以看到 MCP 客户端成功调用了 MCP 服务器提供的 list_indices
从 Elasticsearch 集群中获取到了索引信息。
2025-02-03 17:13:56,518 - mcp.server.lowlevel.server - INFO - Processing request of type ListToolsRequest
Connected to server with tools: ['list_indices', 'get_index', 'write_documents']
MCP Client Started!
Type your queries or 'quit' to exit.
Query: Elasticsearch 集群中有哪些索引?
2025-02-03 17:14:04,537 - mcp.server.lowlevel.server - INFO - Processing request of type ListToolsRequest
2025-02-03 17:14:06,185 - mcp.server.lowlevel.server - INFO - Processing request of type CallToolRequest
2025-02-03 17:14:06,248 - elastic_transport.transport - INFO - GET https://localhost:9200/_cat/indices?format=json [status:200 duration:0.062s]
2025-02-03 17:14:06,248 - mcp.server.lowlevel.server - INFO - Warning: InsecureRequestWarning: Unverified HTTPS request is being made to host 'localhost'. Adding certificate verification is strongly advised. See: https://urllib3.readthedocs.io/en/latest/advanced-usage.html#tls-warnings
[Calling tool list_indices with args {}]
在Elasticsearch集群中有以下索引:
1. 索引名称: student
- 健康状态: green
- 状态: open
- UUID: iKl3j9ujSJ2i0GF5v8NUDw
- 主分片数: 1
- 副本分片数: 1
- 文档数量: 3
- 删除的文档数量: 0
- 存储大小: 12kb
- 主分片存储大小: 6kb
2. 索引名称: teacher
- 健康状态: green
- 状态: open
- UUID: vfbwvH7yQGWqRGEG-t4FnA
- 主分片数: 1
- 副本分片数: 1
- 文档数量: 3
- 删除的文档数量: 0
- 存储大小: 22.6kb
- 主分片存储大小: 11.3kb
3. 索引名称: movies
- 健康状态: green
- 状态: open
- UUID: JYhacHmXREWmkvwXBaFPmg
- 主分片数: 1
- 副本分片数: 1
- 文档数量: 3
- 删除的文档数量: 0
- 存储大小: 13.2kb
- 主分片存储大小: 6.6kb
总结
本文介绍了如何使用 MCP Python SDK 编写一个 MCP 客户端，并集成 LLM 来实现灵活的工具调用和数据处理。通过简单的示例和进阶示例，展示了如何通过标准输入输出（stdio）方式与 MCP 服务器建立连接，并集成 LLM（如通义千问）来实现更复杂的应用场景。
参考资料
For Client Developers：https://modelcontextprotocol.io/quickstart/client https://github.com/modelcontextprotocol/python-sdk：https://github.com/modelcontextprotocol/python-sdk
