AI导哈

MCP客户端库

一个用于通过HTTP与模型上下文协议(MCP)服务器交互的Python客户端库，支持认证、TLS配置、异步操作和场景执行等功能。

  • by TamiTakamiya
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 TamiTakamiya/mcp-client

使用说明

MCP客户端

用于通过HTTP与模型上下文协议（MCP）服务器交互的Python客户端库。 尽管这最初是为测试而写的 AAP MCP服务器, 您应该能够将其用于使用流式HTTP进行传输的其他MCP服务器。

特性

  • 基于HTTP的MCP通信 -使用流式HTTP传输连接到MCP服务器
  • 身份验证支持 -使用API密钥的承载令牌身份验证
  • 可配置的TLS验证 -支持自签名证书和经过验证的TLS
  • 基于类别的路由 -用于组织MCP工具的可选类别端点（用于 AAP MCP服务器 仅）
  • 异步/等待支持 -对现代Python应用程序的完全异步支持
  • 健康检查 -内置健康检查端点支持
  • 场景执行 -使用初始化会话执行自定义MCP方案
  • 工具发现 -列出MCP服务器上的可用工具
  • 适当的资源清理 -自动清理HTTP连接以防止挂起

安装

使用紫外线（推荐）

# Install in development mode
uv pip install -e .

# Or install from source
git clone https://github.com/TamiTakamiya/mcp-client.git
cd mcp-client
uv pip install -e .

使用pip

pip install -e .

快速开始

基本用法

import asyncio
from mcpclient import MCPClient

async def main():
    # Create client (TLS verification disabled for self-signed certs)
    client = MCPClient(
        base_url="https://your-mcp-server.com",
        api_key="your-api-key",
        category="job_management"
    )

    # Check server health
    response = client.health_check()
    print(f"Server status: {response.status_code}")

    # Get available tools
    tools = await client.get_tools()
    for tool in tools:
        print(f"Tool: {tool.name}")

asyncio.run(main())

使用TLS验证

from mcpclient import MCPClient

# Enable TLS verification for production servers
client = MCPClient(
    base_url="https://production-server.com",
    api_key="your-api-key",
    category="job_management",
    verify_tls=True  # Enable certificate verification
)

运行自定义场景

from mcpclient import MCPClient
from mcp import ClientSession

async def my_scenario(session: ClientSession):
    # List available tools
    tools = await session.list_tools()
    print(f"Found {len(tools.tools)} tools")

    # Call a specific tool
    result = await session.call_tool(
        name="controller.job_templates_list",
        arguments={"version": "v2"}
    )

    return result


async def main():
    # Execute the scenario
    client = MCPClient(base_url="https://server.com", api_key="key", category="job_management")
    result = await client.run_a_scenario(my_scenario)
    print(result)

asyncio.run(main())

API 参考

MCP客户端

构造函数

MCPClient(
    base_url: str,
    api_key: str,
    category: str | None = None,
    verify_tls: bool = False
)

参数：

  • base_url （str）：MCP服务器的基本URL（例如，“https://example.com")
  • api_key （str）：承载令牌身份验证的API密钥
  • category （str|None）：端点路由的可选类别
  • verify_tls （bool）：启用TLS证书验证（默认值：False）

方法

health_check() -> httpx.Response

检查服务器运行状况。

response = client.health_check()
assert response.status_code == 200
async get_tools() -> list[types.Tool]

从MCP服务器获取可用工具。

tools = await client.get_tools()
for tool in tools:
    print(f"Tool: {tool.name}")
async run_a_scenario(scenario_func: Callable) -> Any

使用初始化的MCP会话运行自定义场景函数。

async def scenario(session: ClientSession):
    tools = await session.list_tools()
    return tools.tools

result = await client.run_a_scenario(scenario)

配置

环境变量

可以使用环境变量配置客户端：

export SERVER_URL="https://your-mcp-server.com"
export API_KEY="your-api-key"

然后在代码中：

import os
from mcpclient import MCPClient

client = MCPClient(
    base_url=os.environ["SERVER_URL"],
    api_key=os.environ["API_KEY"]
)

TLS配置

对于自签名证书（默认）：

client = MCPClient(base_url=url, api_key=key, verify_tls=False)

对于持有有效证书的生产：

client = MCPClient(base_url=url, api_key=key, verify_tls=True)

测试

运行测试

在开始测试之前，创建 .env 文件定义运行测试所需的环境变量。

API_KEY=your-aap-api-key
SERVER_URL=your-mcp-server-url

该项目使用pytest进行测试：

# Install test dependencies
uv pip install -e ".[dev]"

# Run all tests
uv run --env-file=.env pytest

# Run specific test file
uv run --env-file=.env pytest tests/test_mcpserver.py

# Run with verbose output
uv run --env-file=.env pytest -v

# Run async tests only
uv run --env-file=.env pytest -k "asyncio"

测试配置

测试需要环境变量：

export SERVER_URL="https://test-server.com"
export API_KEY="test-api-key"

如果未设置这些，则将自动跳过测试。

在覆盖范围内运行测试

pytest --cov=mcpclient --cov-report=html

发展

设置开发环境

# Clone the repository
git clone https://github.com/TamiTakamiya/mcp-client.git
cd mcp-client

# Install in development mode with dev dependencies
uv pip install -e ".[dev]"

# Run tests
pytest

项目结构

mcp-client/
├── mcpclient/              # Main package
│   ├── __init__.py         # Package exports
│   └── mcpclient.py        # MCPClient class
├── tests/                  # Test files
│   └── test_mcpserver.py
├── pyproject.toml          # Project configuration
└── README.md               # This file

代码质量

代码库包括：

  • 所有公共API的类型提示
  • 全面的文档字符串
  • 通过Python的日志支持 logging 模块
  • 正确的异步资源清理

日志记录

该库使用Python的标准日志模块：

import asyncio
import logging
from mcpclient import MCPClient

# Enable debug logging
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

async def main():
    # Use the client - will log tool information
    client = MCPClient(base_url="https://server.com", api_key="your-key")
    tools = await client.get_tools()
    print(f"Found {len(tools)} tools")

asyncio.run(main())

故障排除

悬挂测试

如果异步测试挂起，请确保在没有适当清理的情况下创建客户端实例。库现在使用延迟客户端创建来防止此问题。

SSL证书错误

如果遇到SSL证书错误：

  1. 对于自签名证书：使用 verify_tls=False （默认）
  2. 生产：使用 verify_tls=True 并确保证书有效

导入错误

如果您收到导入错误：

# Reinstall in development mode
uv pip install -e .

需求

  • Python>=3.12
  • mcp[cli]>=1.20.0
  • httpx（与mcp一起安装）

许可证

[在此处指定您的许可证]

贡献

[在此处指定捐款指南]

支持

有关问题和疑问，请在GitHub上打开问题。

相关 MCP