从 MCP 服务器生成可发现的代码供 AI 代理工具使用。
模型上下文协议 (MCP) 允许 AI 代理连接到外部工具和数据。然而,正如 Anthropic 的 工程文章 所解释的,直接将数百或数千个工具定义加载到代理的上下文窗口中是低效的:
解决方案:将 MCP 服务器呈现为代理可以通过文件系统接口发现和使用的 代码 API。这种方法可以减少高达 98.7% 的令牌使用量,同时使更强大的代理工作流程成为可能。
问题:手动为每个 MCP 工具编写包装代码是乏味的。
mcp-coded-tools:自动从任何 MCP 服务器生成可发现的 Python 代码。
pip install mcp-coded-tools
# 从流行的 MCP 服务器生成代码
mcp-coded-tools generate \
--command "npx -y @modelcontextprotocol/server-github" \
--output ./servers \
--server-name github
# 从多个服务器生成以完成整个工作流
mcp-coded-tools generate \
--command "npx -y @modelcontextprotocol/server-github" \
--command "npx -y @modelcontextprotocol/server-postgres" \
--command "python slack_mcp_server.py" \
--output ./servers
# 开发模式(更改时自动重新生成)
mcp-coded-tools generate \
--command "python ./my_mcp_server.py" \
--output ./tools \
--watch
💡 参见 POPULAR_MCP_SERVERS.md 获取 50 多个流行服务器和实际用例!
💡 参见 WATCH_MODE.md 获取开发期间的自动重新生成!
import asyncio
from mcp-coded-tools import MCPCodeGenerator
async def main():
generator = MCPCodeGenerator()
# 连接到 MCP 服务器并生成代码
await generator.connect_and_scan([
"npx", "-y", "@modelcontextprotocol/server-gdrive"
])
generator.generate_code(
output_dir="./servers",
server_name="google_drive"
)
print("✓ 生成了可发现的代码!")
asyncio.run(main())
servers/
├── google_drive/
│ ├── __init__.py
│ ├── get_document.py
│ ├── list_files.py
│ └── ...
├── salesforce/
│ ├── __init__.py
│ ├── update_record.py
│ └── ...
└── _client.py
每个工具都成为一个带有类型的 Python 函数:
# servers/google_drive/get_document.py
from typing import Optional, Dict, Any
from .._client import call_mcp_tool
async def get_document(
document_id: str,
fields: Optional[str] = None
) -> Dict[str, Any]:
"""
从 Google Drive 检索文档
"""
return await call_mcp_tool(
'gdrive_getDocument',
{'documentId': document_id, 'fields': fields}
)
代理通过探索文件系统来发现工具:
# 代理列出可用的服务器
import os
servers = os.listdir('./servers')
# ['google_drive', 'salesforce', ...]
# 代理读取特定工具
with open('./servers/google_drive/get_document.py') as f:
tool_def = f.read()
# 理解参数、类型、描述
# 代理编写代码以使用工具
import servers.google_drive as gdrive
import servers.salesforce as sf
async def sync_meeting_notes():
doc = await gdrive.get_document(document_id="abc1_23")
await sf.update_record(
object_type="Lead",
record_id="xyz789",
data={"Notes": doc['content']}
)
默认情况下,工具名称被分析以提取服务器前缀:
gdrive_getDocument → gdrive 服务器salesforce_updateRecord → salesforce 服务器使用显式服务器名称覆盖:
generator.generate_code(
output_dir="./servers",
server_name="my_custom_name"
)
使用 Jinja2 模板自定义生成的代码:
generator = MCPCodeGenerator(
template_dir="./my_templates"
)
跨多个系统自动化 DevOps 工作流:
mcp-coded-tools generate \
--command "npx -y @modelcontextprotocol/server-postgres" \
--command "npx -y @modelcontextprotocol/server-github" \
--command "python slack_mcp_server.py" \
--output ./devops_tools
工作流:查询错误 → 创建问题 → 通知团队 令牌节省:98%(150K → 3K 令牌) 成本影响:每次事件从 $3.00 降低到 $0.06
处理数百万行数据而不污染上下文:
from data_tools.postgres import execute_query
from data_tools.slack import post_message
# 查询 1M 行 - 保持在执行环境中!
rows = await execute_query(
query="SELECT * FROM transactions WHERE date > NOW() - INTERVAL '30 days'"
)
# 在代码中处理数据(从未进入上下文)
import pandas as pd
df = pd.DataFrame(rows)
summary = df.groupby('user_id')['amount'].sum()
# 只有摘要进入上下文
await post_message(
channel='analytics',
text=f"处理了 {len(df):,} 笔交易,总计:${summary.sum():,.2f}"
)
规模:处理 TB 级数据,而不是 MB 级 投资回报率:每年 $109M 的收益,每天运行 500 次查询的操作
大规模地使用 AI 辅助审查 PR:
mcp-coded-tools generate \
--command "npx -y @modelcontextprotocol/server-github" \
--command "npx -y @modelcontextprotocol/server-filesystem" \
--command "npx -y @modelcontextprotocol/server-sequential-thinking" \
--output ./code_review_tools
影响:每天审查数千个 PR,并保持一致的质量
📚 参见 examples/real_world_workflows.py 获取 5 个完整的生产示例!
# 克隆仓库
git clone https://github.com/bluman1/mcp-coded-tools.git
cd mcp-coded-tools
# 以开发模式安装
pip install -e ".[dev]"
# 运行测试
pytest
# 格式化代码
black src/ tests/
ruff check src/ tests/
# 类型检查
mypy src/
欢迎贡献!请阅读 CONTRIBUTING.md 了解指南。
MIT 许可证 - 详情见 LICENSE。
灵感来自 Anthropic 的 关于使用 MCP 执行代码的工程文章。
由 Michael Ogundare 为 MCP 社区构建。