03-MCP客户端开发实战

概述

在前两篇文章中,我们分别介绍了 MCP 协议的基本概念和如何构建 MCP 服务器。本文将详细介绍如何开发 MCP 客户端,包括 stdio 和 HTTP 两种传输方式的实现,以及如何调用工具、读取资源和获取提示模板。

MCP客户端架构

客户端职责

MCP 客户端负责与 MCP 服务器建立连接,并管理与服务器的交互:

• 建立和维护服务器连接
• 发现服务器提供的工具、资源和提示模板
• 调用工具并处理返回结果
• 读取资源并解析数据
• 获取提示模板并应用

核心组件

1. 传输层: stdio 或 HTTP 连接管理
2. 会话层: ClientSession 管理
3. API 层: 工具调用、资源读取、提示模板获取
4. 错误处理: 异常捕获和重试机制

Stdio模式客户端

基础结构

Stdio 模式通过标准输入输出与服务器进程通信。

import asyncio
from mcp.client.stdio import stdio_client
from mcp import ClientSession, StdioServerParameters

async def main():
    # 配置服务器参数
    server_params = StdioServerParameters(
        command="python",
        args=["math_mcp_server_stdio.py"]
    )

    # 建立连接
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

if __name__ == "__main__":
    asyncio.run(main())

服务器参数配置

StdioServerParameters 用于配置 stdio 连接:

server_params = StdioServerParameters(
    command="python",           # 执行命令
    args=["math_mcp_server_stdio.py"],  # 命令参数
    env=None,                  # 环境变量(可选)
    cwd=None                   # 工作目录(可选)
)

会话初始化

async with stdio_client(server_params) as (read, write):
    async with ClientSession(read, write) as session:
        # 初始化会话
        capabilities = await session.initialize()

        # 检查服务器能力
        print(f"Server capabilities: {capabilities}")

完整的 Stdio 客户端

以下是完整的 Stdio 客户端实现:

import asyncio
from mcp.client.stdio import stdio_client
from mcp import ClientSession, StdioServerParameters

async def main():
    # 配置服务器参数
    server_params = StdioServerParameters(
        command="python",
        args=["math_mcp_server_stdio.py"]
    )

    print("=" * 70)
    print("MCP数学计算服务器客户端演示")
    print("MCP Math Server Client Demo")
    print("=" * 70)
    print()

    # 连接到服务器
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # 初始化会话
            await session.initialize()
            print("成功连接到服务器!")
            print("Successfully connected to server!")
            print()

            # 第1部分: 列出所有可用工具
            print("=" * 70)
            print("第1部分: 可用工具列表")
            print("Part 1: Available Tools")
            print("=" * 70)
            tools = await session.list_tools()
            print(f"\n找到 {len(tools.tools)} 个工具 (Found {len(tools.tools)} tools):")
            print()
            for idx, tool in enumerate(tools.tools, 1):
                print(f"{idx}. {tool.name}")
                print(f"   描述 (Description): {tool.description}")
            print()

            # 第2部分: 调用基础运算工具
            print("=" * 70)
            print("第2部分: 基础运算示例")
            print("Part 2: Basic Arithmetic Examples")
            print("=" * 70)
            print()

            # 加法
            print("调用加法工具: add(10, 20)")
            result = await session.call_tool("add", {"a": 10, "b": 20})
            print(f"结果 (Result): {result.content[0].text}")
            print()

            # 减法
            print("调用减法工具: subtract(50, 30)")
            result = await session.call_tool("subtract", {"a": 50, "b": 30})
            print(f"结果 (Result): {result.content[0].text}")
            print()

            # 乘法
            print("调用乘法工具: multiply(7, 8)")
            result = await session.call_tool("multiply", {"a": 7, "b": 8})
            print(f"结果 (Result): {result.content[0].text}")
            print()

            # 除法
            print("调用除法工具: divide(100, 4)")
            result = await session.call_tool("divide", {"a": 100, "b": 4})
            print(f"结果 (Result): {result.content[0].text}")
            print()

            # 幂运算
            print("调用幂运算工具: power(2, 10)")
            result = await session.call_tool("power", {"base": 2, "exponent": 10})
            print(f"结果 (Result): {result.content[0].text}")
            print()

            # 平方根
            print("调用平方根工具: sqrt(144)")
            result = await session.call_tool("sqrt", {"number": 144})
            print(f"结果 (Result): {result.content[0].text}")
            print()

            # 阶乘
            print("调用阶乘工具: factorial(5)")
            result = await session.call_tool("factorial", {"n": 5})
            print(f"结果 (Result): {result.content[0].text}")
            print()

            # 第3部分: 列出所有可用资源
            print("=" * 70)
            print("第3部分: 可用资源列表")
            print("Part 3: Available Resources")
            print("=" * 70)
            resources = await session.list_resources()
            print(f"\n找到 {len(resources.resources)} 个资源 (Found {len(resources.resources)} resources):")
            print()
            for idx, resource in enumerate(resources.resources, 1):
                print(f"{idx}. {resource.uri}")
                print(f"   名称 (Name): {resource.name}")
            print()

            # 第4部分: 读取各种资源
            print("=" * 70)
            print("第4部分: 读取资源示例")
            print("Part 4: Read Resources Examples")
            print("=" * 70)
            print()

            # 读取圆周率
            print("读取圆周率资源: constant://pi")
            resource = await session.read_resource("constant://pi")
            print(f"圆周率值 (Pi value): {resource.contents[0].text}")
            print()

            # 读取自然常数
            print("读取自然常数资源: constant://e")
            resource = await session.read_resource("constant://e")
            print(f"自然常数 (Euler's number): {resource.contents[0].text}")
            print()

            # 读取黄金比例
            print("读取黄金比例资源: constant://golden_ratio")
            resource = await session.read_resource("constant://golden_ratio")
            print(f"黄金比例 (Golden ratio): {resource.contents[0].text}")
            print()

            # 读取个性化问候
            print("读取个性化问候: greeting://Alice")
            resource = await session.read_resource("greeting://Alice")
            print(f"问候 (Greeting): {resource.contents[0].text}")
            print()

            # 读取配置
            print("读取应用配置: config://app")
            resource = await session.read_resource("config://app")
            print("配置信息 (Configuration):")
            for line in resource.contents[0].text.split('\n'):
                print(f"  {line}")
            print()

            # 第5部分: 列出所有提示模板
            print("=" * 70)
            print("第5部分: 可用提示模板列表")
            print("Part 5: Available Prompts")
            print("=" * 70)
            prompts = await session.list_prompts()
            print(f"\n找到 {len(prompts.prompts)} 个提示模板 (Found {len(prompts.prompts)} prompts):")
            print()
            for idx, prompt in enumerate(prompts.prompts, 1):
                print(f"{idx}. {prompt.name}")
                print(f"   描述 (Description): {prompt.description}")
            print()

            # 第6部分: 获取提示模板
            print("=" * 70)
            print("第6部分: 获取提示模板示例")
            print("Part 6: Get Prompt Examples")
            print("=" * 70)
            print()

            # 获取数学问题提示
            print("获取数学问题提示: example_prompt(什么是1+1?)")
            prompt = await session.get_prompt("example_prompt", {"question": "什么是1+1?"})
            print("提示内容 (Prompt content):")
            for line in prompt.messages[0].content.text.split('\n'):
                print(f"  {line}")
            print()

            # 获取计算指导
            print("获取乘法指导: calculation_guide(multiplication)")
            prompt = await session.get_prompt("calculation_guide", {"operation": "multiplication"})
            print("指导内容 (Guide content):")
            for line in prompt.messages[0].content.text.split('\n'):
                print(f"  {line}")
            print()

            print("=" * 70)
            print("演示完成!")
            print("Demo completed!")
            print("=" * 70)
            print()
            print("总结 (Summary):")
            print("  - 成功连接到MCP服务器")
            print("  - 列出了所有可用的工具、资源和提示模板")
            print("  - 调用了7个不同的数学运算工具")
            print("  - 读取了6个不同的资源")
            print("  - 获取了2个提示模板")
            print()

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\n\n程序被用户中断 (Program interrupted by user)")
    except Exception as e:
        print(f"\n\n发生错误 (Error occurred): {e}")
        import traceback
        traceback.print_exc()

运行 Stdio 客户端

# 确保服务器正在运行
python math_mcp_server_stdio.py

# 在另一个终端运行客户端
python math_mcp_client_stdio.py

HTTP模式客户端

基础结构

HTTP 模式通过 HTTP 协议与服务器通信。

import asyncio
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

async def main():
    # 服务器 URL
    server_url = "http://localhost:8000/mcp"

    # 建立连接
    async with streamablehttp_client(server_url) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()

if __name__ == "__main__":
    asyncio.run(main())

完整的 HTTP 客户端

import asyncio
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

# 数学HTTP服务器URL
math_server_url = "http://localhost:8000/mcp"

async def main():
    print("=" * 70)
    print("MCP数学HTTP服务器客户端演示")
    print("MCP Math HTTP Server Client Demo")
    print("=" * 70)
    print()

    async with streamablehttp_client(math_server_url) as (read, write, _):
        async with ClientSession(read, write) as session:
            # 初始化会话
            await session.initialize()
            print("成功连接到数学HTTP服务器!")
            print("Successfully connected to Math HTTP server!")
            print()

            # 列出所有可用工具
            print("=" * 70)
            print("第1部分: 可用工具列表")
            print("Part 1: Available Tools")
            print("=" * 70)
            response_tools = await session.list_tools()
            print(f"\n找到 {len(response_tools.tools)} 个工具 (Found {len(response_tools.tools)} tools):")
            print()
            for idx, tool in enumerate(response_tools.tools, 1):
                print(f"{idx}. 工具名称 (Name): {tool.name}")
                print(f"   描述 (Description): {tool.description}")
            print()

            # 列出所有资源
            print("=" * 70)
            print("第2部分: 可用资源列表")
            print("Part 2: Available Resources")
            print("=" * 70)
            response_resources = await session.list_resources()
            print(f"\n找到 {len(response_resources.resources)} 个资源 (Found {len(response_resources.resources)} resources):")
            print()
            for idx, resource in enumerate(response_resources.resources, 1):
                print(f"{idx}. 资源URI (URI): {resource.uri}")
                print(f"   资源名称 (Name): {resource.name}")
            print()

            # 列出所有提示模板
            print("=" * 70)
            print("第3部分: 可用提示模板列表")
            print("Part 3: Available Prompts")
            print("=" * 70)
            response_prompts = await session.list_prompts()
            print(f"\n找到 {len(response_prompts.prompts)} 个提示模板 (Found {len(response_prompts.prompts)} prompts):")
            print()
            for idx, prompt in enumerate(response_prompts.prompts, 1):
                print(f"{idx}. 提示名称 (Name): {prompt.name}")
                print(f"   描述 (Description): {prompt.description}")
            print()

            # 调用工具示例
            print("=" * 70)
            print("第4部分: 调用工具示例")
            print("Part 4: Tool Call Examples")
            print("=" * 70)
            print()

            # 示例1: 加法
            print("调用加法工具: add(10, 20)")
            result = await session.call_tool("add", arguments={"a": 10, "b": 20})
            print(f"结果 (Result): {result.content[0].text}")
            print()

            # 示例2: 乘法
            print("调用乘法工具: multiply(7, 8)")
            result = await session.call_tool("multiply", arguments={"a": 7, "b": 8})
            print(f"结果 (Result): {result.content[0].text}")
            print()

            # 示例3: 幂运算
            print("调用幂运算工具: power(2, 10)")
            result = await session.call_tool("power", arguments={"base": 2, "exponent": 10})
            print(f"结果 (Result): {result.content[0].text}")
            print()

            # 示例4: 平方根
            print("调用平方根工具: sqrt(144)")
            result = await session.call_tool("sqrt", arguments={"number": 144})
            print(f"结果 (Result): {result.content[0].text}")
            print()

            # 读取资源示例
            print("=" * 70)
            print("第5部分: 读取资源示例")
            print("Part 5: Read Resource Examples")
            print("=" * 70)
            print()

            # 示例5: 读取圆周率
            print("读取圆周率资源: constant://pi")
            resource = await session.read_resource("constant://pi")
            print(f"圆周率值 (Pi value): {resource.contents[0].text}")
            print()

            # 示例6: 读取自然常数
            print("读取自然常数资源: constant://e")
            resource = await session.read_resource("constant://e")
            print(f"自然常数 (Euler's number): {resource.contents[0].text}")
            print()

            # 获取提示模板示例
            print("=" * 70)
            print("第6部分: 获取提示模板示例")
            print("Part 6: Get Prompt Examples")
            print("=" * 70)
            print()

            # 示例7: 获取数学问题提示
            print("获取数学问题提示: example_prompt(什么是10+2?)")
            prompt = await session.get_prompt("example_prompt", arguments={"question": "What is 10+2?"})
            print(f"提示内容 (Prompt content):")
            print(f"  {prompt.messages[0].content.text}")
            print()

            print("=" * 70)
            print("演示完成!")
            print("Demo completed!")
            print("=" * 70)

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\n\n程序被用户中断 (Program interrupted by user)")
    except Exception as e:
        print(f"\n\n发生错误 (Error occurred): {e}")
        import traceback
        traceback.print_exc()

运行 HTTP 客户端

# 终端1: 启动 HTTP 服务器
python math_mcp_server_http.py

# 终端2: 运行客户端
python math_mcp_client_http.py

核心API详解

1. 工具调用

列出工具

# 获取所有可用工具
tools = await session.list_tools()

# 遍历工具列表
for tool in tools.tools:
    print(f"工具名称: {tool.name}")
    print(f"工具描述: {tool.description}")
    print(f"输入Schema: {tool.inputSchema}")

调用工具

# 调用加法工具
result = await session.call_tool("add", {"a": 10, "b": 20})

# 获取结果
for content in result.content:
    if hasattr(content, 'text'):
        print(f"结果: {content.text}")

错误处理

try:
    result = await session.call_tool("divide", {"a": 10, "b": 0})
except Exception as e:
    print(f"工具调用失败: {e}")

2. 资源读取

列出资源

# 获取所有可用资源
resources = await session.list_resources()

# 遍历资源列表
for resource in resources.resources:
    print(f"资源URI: {resource.uri}")
    print(f"资源名称: {resource.name}")
    print(f"资源描述: {resource.description}")

读取资源

# 读取静态资源
resource = await session.read_resource("constant://pi")
print(f"圆周率: {resource.contents[0].text}")

# 读取动态资源
resource = await session.read_resource("greeting://Alice")
print(f"问候: {resource.contents[0].text}")

# 读取JSON资源
resource = await session.read_resource("config://app")
import json
config = json.loads(resource.contents[0].text)
print(f"配置: {config}")

3. 提示模板获取

列出提示模板

# 获取所有可用提示模板
prompts = await session.list_prompts()

# 遍历提示模板列表
for prompt in prompts.prompts:
    print(f"提示名称: {prompt.name}")
    print(f"提示描述: {prompt.description}")
    print(f"参数: {prompt.arguments}")

获取提示模板

# 获取系统提示
prompt = await session.get_prompt("system_prompt")
print(f"系统提示: {prompt.messages[0].content.text}")

# 获取带参数的提示
prompt = await session.get_prompt("example_prompt", {"question": "What is 1+1?"})
print(f"提示: {prompt.messages[0].content.text}")

高级功能

1. 并发调用

async def concurrent_calls():
    tasks = [
        session.call_tool("add", {"a": 1, "b": 2}),
        session.call_tool("multiply", {"a": 3, "b": 4}),
        session.call_tool("divide", {"a": 10, "b": 2})
    ]

    results = await asyncio.gather(*tasks)

    for i, result in enumerate(results):
        print(f"结果{i+1}: {result.content[0].text}")

asyncio.run(concurrent_calls())

2. 重试机制

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10))
async def safe_call_tool(session, tool_name, arguments):
    return await session.call_tool(tool_name, arguments)

3. 日志记录

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

async def logged_call(session, tool_name, arguments):
    logger.info(f"调用工具: {tool_name} 参数: {arguments}")
    try:
        result = await session.call_tool(tool_name, arguments)
        logger.info(f"工具调用成功: {result.content[0].text}")
        return result
    except Exception as e:
        logger.error(f"工具调用失败: {e}")
        raise

4. 连接池管理

class MCPClientPool:
    def __init__(self, server_url, max_connections=5):
        self.server_url = server_url
        self.max_connections = max_connections
        self.clients = []

    async def get_client(self):
        if len(self.clients) < self.max_connections:
            read, write, _ = await streamablehttp_client(self.server_url).__aenter__()
            client = ClientSession(read, write)
            await client.__aenter__()
            await client.initialize()
            self.clients.append(client)
        return self.clients[-1]

    async def close_all(self):
        for client in self.clients:
            await client.__aexit__(None, None, None)
        self.clients.clear()

客户端最佳实践

1. 资源管理

# 使用 async with 确保资源释放
async with stdio_client(server_params) as (read, write):
    async with ClientSession(read, write) as session:
        # 客户端逻辑
        pass
# 自动释放资源

2. 错误处理

async def robust_client():
    try:
        async with stdio_client(server_params) as (read, write):
            async with ClientSession(read, write) as session:
                await session.initialize()
                # 客户端逻辑
    except ConnectionError as e:
        print(f"连接失败: {e}")
    except TimeoutError:
        print("请求超时")
    except Exception as e:
        print(f"未知错误: {e}")

3. 配置管理

import os
from dataclasses import dataclass

@dataclass
class ClientConfig:
    server_url: str
    timeout: int = 30
    max_retries: int = 3
    log_level: str = "INFO"

def load_config():
    return ClientConfig(
        server_url=os.getenv("MCP_SERVER_URL", "http://localhost:8000/mcp"),
        timeout=int(os.getenv("MCP_TIMEOUT", "30")),
        max_retries=int(os.getenv("MCP_MAX_RETRIES", "3"))
    )

4. 性能优化

import functools

def cache_tool_call(ttl=60):
    """工具调用结果缓存装饰器"""
    cache = {}

    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            key = str(args) + str(kwargs)
            if key in cache:
                return cache[key]
            result = await func(*args, **kwargs)
            cache[key] = result
            return result
        return wrapper
    return decorator

测试客户端

单元测试

import pytest
from unittest.mock import AsyncMock, patch

@pytest.mark.asyncio
async def test_call_tool():
    async with patch('mcp.client.stdio.stdio_client') as mock_client:
        mock_session = AsyncMock()
        mock_session.initialize = AsyncMock()
        mock_session.call_tool = AsyncMock(return_value=Mock(content=[Mock(text="30")]))

        mock_client.return_value.__aenter__.return_value = (None, None)
        mock_client.return_value.__aexit__.return_value = None

        async with stdio_client(None) as (read, write):
            async with ClientSession(read, write) as session:
                result = await session.call_tool("add", {"a": 10, "b": 20})
                assert result.content[0].text == "30"

集成测试

@pytest.mark.asyncio
async def test_math_server_integration():
    server_params = StdioServerParameters(
        command="python",
        args=["math_mcp_server_stdio.py"]
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # 测试工具调用
            result = await session.call_tool("add", {"a": 10, "b": 20})
            assert "30" in result.content[0].text

            # 测试资源读取
            resource = await session.read_resource("constant://pi")
            assert "3.14" in resource.contents[0].text

故障排查

问题 1: 连接失败

错误信息:

ConnectionError: Unable to connect to server

解决方案:

• 检查服务器是否运行
• 验证服务器URL是否正确
• 检查防火墙设置

问题 2: 工具调用超时

错误信息:

TimeoutError: Tool call timed out

解决方案:

• 增加超时时间
• 优化工具执行效率
• 使用并发调用

问题 3: 参数验证失败

错误信息:

ValidationError: Invalid arguments for tool 'add'

解决方案:

• 检查参数类型
• 验证参数值范围
• 查看工具Schema定义

阅读顺序建议

1. 01-MCP协议入门指南: 了解 MCP 基本概念和核心组件
2. 02-快速构建MCP服务器: 使用 FastMCP 构建服务器
3. 03-MCP客户端开发实战: 开发 stdio 和 HTTP 客户端
4. 04-LLM与MCP集成实践: 集成到 LangGraph 构建智能代理
5. 05-多服务器架构与最佳实践: 多服务器架构和生产部署

总结

本文详细介绍了 MCP 客户端的开发,包括:

• Stdio 模式客户端的完整实现
• HTTP 模式客户端的完整实现
• 核心 API 的详细使用方法
• 高级功能和最佳实践
• 测试和故障排查

在下一篇《LLM与MCP集成实践》中,我们将学习如何将 MCP 工具集成到 LangGraph 中,构建智能代理应用。

参考资源

• MCP SDK 文档
• AsyncIO 官方文档
• Python 类型注解

文章标签

MCP客户端, Python开发, stdio传输, HTTP客户端, 工具调用, 资源读取, 提示模板, 异步编程, 客户端开发