使用说明
简单MCP客户端
一个干净、最小的MCP（模型上下文协议）客户端，便于集成到任何应用程序中。此库提供了连接到MCP服务器和执行工具的核心功能，同时将Interactive CLI单独用于测试目的。
🚀 快速开始
安装
# Install dependencies
uv sync
# Or with pip
pip install -e .
基本用法
from simple_mcp_client import SimpleMCPClient
async def example():
# Create client
async with SimpleMCPClient() as client:
# Connect to server
tools = await client.connect_to_server("my-server")
# Execute a tool
result = await client.execute_tool("my_tool", {"arg": "value"})
print(result)
📁 项目结构
simple-mcp-client/
├── simple_mcp_client.py # 🎯 Main MCP client (for developers)
├── interactive_cli.py # 🖥️ Interactive CLI (for testing)
├── llm_providers.py # 🧠 LLM provider implementations
├── config.py # ⚙️ Configuration management
├── query_processing.py # 🔄 Query processing utilities & multi-server support
├── mcp_client.py # 🔧 Legacy compatibility module
├── mcp_config_example.json # 📡 MCP server configuration template
├── mcp_config.json # 📡 Your local MCP config (gitignored)
├── llm_config.json # 🤖 Your local LLM config (gitignored)
├── example_configs/ # 📋 Configuration examples
└── examples/ # 📚 Integration examples
├── basic_usage.py
├── web_app_integration.py
└── ai_agent_integration.py
🎯 开发人员：使用SimpleMCPClient
这 SimpleMCPClient 旨在成为 主界面 对于希望将MCP功能集成到其应用程序中的开发人员。
🔄 最近重构：代码库已经过简化，以提高可维护性：
- 消除了冗余代码：删除了约250行重复功能
- 整合架构：公用设施已移至
query_processing.py - 改进的多服务器支持：增强了对多个MCP服务器的支持
- 更好的测试：更集中、更全面的测试覆盖
- 保持向后兼容性：所有现有API继续工作
主要特点
- 清洁API：MCP操作的简单直观方法
- 多服务器支持：同时连接到多个MCP服务器
- 异步支持：具有上下文管理器的完全异步/等待支持
- 错误处理：全面的错误处理，有明确的异常
- 工具验证：自动工具验证和有用的错误消息
- 连接管理：自动连接生命周期管理
- FastMCP v2兼容：基于FastMCP v2构建，实现最佳性能
- 无依赖关系：对CLI或LLM组件零依赖
api参考
SimpleMCPClient
class SimpleMCPClient:
def __init__(self, config_path: str = "mcp_config.json", config: Optional[MCPConfig] = None)
# Server Management
def list_servers(self) -> List[str]
def get_server_info(self, server_name: str) -> Dict[str, Any]
# Connection Management
async def connect_to_server(self, server_name: str = None, ...) -> List[Dict[str, Any]]
async def connect_to_all_servers(self) -> Dict[str, List[Dict[str, Any]]]
def is_connected(self) -> bool
def get_current_server(self) -> Optional[str]
def get_connected_servers(self) -> List[str]
# Tool Operations
def get_available_tools(self) -> List[Dict[str, Any]]
def get_tools_by_server(self, server_name: str) -> List[Dict[str, Any]]
async def execute_tool(self, tool_name: str, arguments: Dict[str, Any]) -> str
async def execute_tool_on_server(self, server_name: str, tool_name: str, arguments: Dict[str, Any]) -> str
# Cleanup
async def cleanup(self) -> None
# Context Manager Support
async def __aenter__(self) -> SimpleMCPClient
async def __aexit__(self, exc_type, exc_val, exc_tb) -> None
便利功能
# Execute a single tool
async def execute_mcp_tool(server_name: str, tool_name: str, arguments: Dict[str, Any]) -> str
# List available tools
async def list_mcp_tools(server_name: str) -> List[Dict[str, Any]]
使用示例
1.基本用法
from simple_mcp_client import SimpleMCPClient
async def basic_example():
client = SimpleMCPClient()
try:
# Connect to server
tools = await client.connect_to_server("filesystem")
print(f"Available tools: {[tool['name'] for tool in tools]}")
# Execute tool
result = await client.execute_tool("read_file", {"path": "example.txt"})
print(f"File contents: {result}")
finally:
await client.cleanup()
2.上下文管理器（推荐）
async def context_manager_example():
async with SimpleMCPClient() as client:
tools = await client.connect_to_server("my-server")
result = await client.execute_tool("my_tool", {"param": "value"})
return result
3.直接连接（无配置文件）
async def direct_connection_example():
async with SimpleMCPClient() as client:
tools = await client.connect_to_server(
server_command="python",
server_args=["-m", "my_mcp_server"],
server_env={"CUSTOM_VAR": "value"}
)
# Use tools...
4.错误处理
async def error_handling_example():
async with SimpleMCPClient() as client:
try:
await client.connect_to_server("non-existent-server")
except ValueError as e:
print(f"Configuration error: {e}")
except RuntimeError as e:
print(f"Connection error: {e}")
try:
result = await client.execute_tool("invalid_tool", {})
except ValueError as e:
print(f"Tool not found: {e}")
except RuntimeError as e:
print(f"Execution error: {e}")
5.多服务器支持
async def multi_server_example():
async with SimpleMCPClient() as client:
# Connect to all servers at once
server_tools = await client.connect_to_all_servers()
# Print tools by server
for server, tools in server_tools.items():
print(f"{server}: {[tool['name'] for tool in tools]}")
# Get all available tools across servers
all_tools = client.get_available_tools()
print(f"Total tools: {len(all_tools)}")
# Execute tool on specific server
result = await client.execute_tool_on_server(
"filesystem", "read_file", {"path": "example.txt"}
)
# Or let the client find the tool automatically
result = await client.execute_tool("read_file", {"path": "example.txt"})
6.服务器管理
async def server_management_example():
client = SimpleMCPClient()
# List available servers
servers = client.list_servers()
print(f"Available servers: {servers}")
# Get server information
for server in servers:
info = client.get_server_info(server)
print(f"{server}: {info}")
# Connect to each server
for server in servers:
try:
tools = await client.connect_to_server(server)
print(f"{server} has {len(tools)} tools")
except Exception as e:
print(f"Failed to connect to {server}: {e}")
await client.cleanup()
🖥️ 用于测试：交互式CLI
这 interactive_cli.py 提供a 独立测试界面 它将MCP客户端与LLM提供商相结合，用于交互式测试。
运行CLI
# Set your API key
export ANTHROPIC_API_KEY=your_key_here
# or
export OPENAI_API_KEY=your_key_here
# Run the interactive CLI
python interactive_cli.py
CLI功能
- 自动检测：自动从环境中检测LLM提供程序
- 多供应商：支持Anthropic、OpenAI和OpenRouter
- 多服务器：同时连接到多个MCP服务器
- 交互式聊天：与MCP工具的自然语言交互
- 迭代工具执行：支持复杂工作流的顺序工具调用
- 工具执行：基于LLM决策的自动工具执行
- 服务器管理：列出、连接和切换不同的MCP服务器
- 调试：详细记录发展情况
CLI命令
/help-显示可用命令/tools-列出所有连接服务器中可用的MCP工具/status-显示连接和配置状态/servers-列出可用服务器及其连接状态/connect <server>-连接到特定的MCP服务器/reconnect-重新连接到所有已配置的服务器quit/exit/q-退出CLI
⚙️ 配置
初始设置
重要：出于安全原因，存储库中不包括包含敏感信息的配置文件。
-
复制示例配置:
cp mcp_config_example.json mcp_config.json -
编辑本地配置:
- 用实际项目路径替换占位符路径
- 使用您的凭据更新数据库连接字符串
- 根据您的环境需要修改服务器命令
-
你的
mcp_config.json被忽视了 以防止意外泄露敏感信息。
MCP服务器配置(mcp_config.json)
创建您的本地 mcp_config.json 基于 mcp_config_example.json:
{
"mcpServers": {
"filesystem": {
"command": "python",
"args": ["-m", "mcp_server_filesystem"],
"description": "File system operations",
"env": {
"CUSTOM_VAR": "value"
}
},
"web": {
"command": "npx",
"args": ["-y", "@anthropic/mcp-server-web"],
"description": "Web scraping and requests"
},
"postgres": {
"command": "npx",
"args": [
"-y",
"@modelcontextprotocol/server-postgres",
"postgresql://username:password@localhost:5432/database_name"
],
"description": "PostgreSQL database server"
}
}
}
LLM配置(llm_config.json)-仅限CLI
同样，创建您的本地 llm_config.json （也被忽略了）：
{
"provider": "anthropic",
"model": "claude-3-5-sonnet-20241022",
"max_tokens": 1000,
"options": {}
}
安全说明
⚠️ 从不提交包含以下内容的配置文件:
- API密钥或令牌
- 数据库凭据
- 包含个人信息的绝对文件路径
- 任何敏感的环境变量
示例文件提供了可以在本地自定义的安全模板。
📚 集成示例
WEB应用集成
from simple_mcp_client import SimpleMCPClient
from fastapi import FastAPI
app = FastAPI()
mcp_client = SimpleMCPClient()
@app.post("/execute-tool")
async def execute_tool(server: str, tool: str, args: dict):
async with SimpleMCPClient() as client:
await client.connect_to_server(server)
result = await client.execute_tool(tool, args)
return {"result": result}
AI代理集成
from simple_mcp_client import SimpleMCPClient
from llm_providers import create_llm_provider, LLMProvider
class MCPAgent:
def __init__(self, llm_provider, mcp_client):
self.llm_provider = llm_provider
self.mcp_client = mcp_client
async def process_task(self, task: str) -> str:
# Use LLM to determine which tools to use
# Execute tools via mcp_client
# Return processed result
pass
批处理
async def batch_process_files(file_paths: List[str]):
async with SimpleMCPClient() as client:
await client.connect_to_server("filesystem")
results = []
for path in file_paths:
result = await client.execute_tool("read_file", {"path": path})
results.append(result)
return results
🔧 发展
项目设置
# Clone repository
git clone <repository-url>
cd simple-mcp-client
# Install dependencies
uv sync
# Install with development dependencies
uv sync --dev
代码质量
# Format code
black *.py
# Type checking
mypy simple_mcp_client.py
# Run tests
pytest
测试
# Test basic functionality
python -c "from simple_mcp_client import SimpleMCPClient; print('Import successful')"
# Run examples
python examples/basic_usage.py
# Test interactive CLI
python interactive_cli.py
🌟 关键利益
对于开发者
- 清洁API：简单直观的界面，专注于MCP功能
- 多服务器支持：同时连接到多个MCP服务器
- 零依赖：核心客户端中没有CLI或LLM依赖关系
- FastMCP v2：基于最新的FastMCP v2构建，实现最佳性能
- 异步支持：完全异步/等待支持，具有适当的资源管理
- 错误处理：清除错误消息和正确的异常处理
- 柔性连接：支持基于配置的连接和直接连接
- 上下文管理器：使用异步上下文管理器自动清理
用于测试
- 单独的CLI：独立于核心客户端的交互式测试界面
- LLM集成：与MCP工具的自然语言交互
- 多服务器支持：同时测试多个MCP服务器
- 迭代工具执行：支持复杂的顺序工作流
- 多供应商：支持不同的法学硕士提供者
- 调试：用于开发和故障排除的详细日志记录
📦 依赖项
核心客户(simple_mcp_client.py)
fastmcp>=2.0.0-FastMCP v2协议实现python-dotenv>=1.0.0-环境变量加载
交互式CLI(interactive_cli.py)
anthropic>=0.8.0-API人类克劳德openai>=1.0.0-OpenAI API（可选）
发展
pytest>=7.0.0-测试框架pytest-asyncio>=0.21.0-异步测试支持black>=23.0.0-代码格式ruff>=0.1.0-快速Python lintermypy>=1.0.0-类型检查
🚀 入门指南
- 安装依赖项:
uv sync - 配置MCP服务器：编辑
mcp_config.json - 用于测试：设置API密钥并运行
python interactive_cli.py - 为了发展：进口
SimpleMCPClient开始建设！
📝 例子
看看 examples/ 综合集成示例目录：
basic_usage.py-MCP客户端的基本使用模式web_app_integration.py-Web应用程序与FastAPI的集成ai_agent_integration.py-集成LLM的AI代理开发
🤝 贡献
- 分叉存储库
- 创建要素分支
- 进行更改
- 运行测试和梳理
- 提交拉取请求
📄 许可证
MIT许可证-有关详细信息，请参阅许可证文件
