AI导哈

Python运行时工具执行

MCP Code Execution - Python Runtime 是一个为AI代理设计的Python运行时环境，通过渐进式工具发现和懒加载技术，显著减少与MCP服务器交互时的令牌使用量，适用于高效数据检索和处理场景。

  • by ipdelete
  • Python
  • 本地部署
  • 20 星标
  • 更新于
在 GitHub 查看 ipdelete/mcp-code-execution

使用说明

MCP代码执行-Python运行时

代币减少98.7% 通过模型上下文协议（MCP）服务器的渐进式工具发现。

概述

此运行时使AI代理能够通过渐进式披露模式与MCP工具协同工作：

  1. 代理探索 ./servers/ 发现可用工具
  2. 代理仅读取所需的工具定义
  3. 代理编写Python脚本，通过MCP工具获取数据
  4. 脚本返回结果（原始或已处理）-然后代理可以在后续轮次中处理/汇总

结果发送给代理的令牌减少了约98.7%。

特性

  • 🦥 延迟加载：服务器仅在调用工具时连接
  • 🔒 类型安全：所有工具输入/输出的Pydantic模型
  • 🔄 防御性编码：处理可变MCP响应结构
  • 📦 自动生成的包装：从MCP模式中键入Python函数
  • 🛠️ 现场标准化：处理不一致的API大小写（例如ADO）

安装

先决条件

  • Python 3.11+
  • 紫外线 包管理器
  • Node.js（用于MCP服务器）

设置

# Install uv
curl -LsSf https://astral.sh/uv/install.sh | sh

# Clone repository
git clone https://github.com/ipdelete/mcp-code-execution.git
cd mcp-code-execution

# Install dependencies
uv sync --all-extras

# Install in editable mode
uv pip install -e ".[dev]"

快速开始

1.配置MCP服务器

创建 mcp_config.json:

{
  "mcpServers": {
    "git": {
      "command": "uvx",
      "args": ["mcp-server-git", "--repository", "."]
    },
    "fetch": {
      "command": "uvx",
      "args": ["mcp-server-fetch"]
    }
  }
}

2.生成工具包装和发现模式（可选）

# Generate wrappers from inputSchema
uv run mcp-generate

如果您的MCP服务器丢失 outputSchema 定义，自动生成它们：

# Step 1: Generate discovery config with LLM-powered test parameters
uv run mcp-generate-discovery

# Step 2: Review discovery_config.json and remove/modify as needed

# Step 3: Execute safe tools and infer schemas
uv run mcp-discover

这将在中创建键入的包装器 servers/:

servers/
  git/
    __init__.py
    git_status.py
    git_log.py
    git_diff.py
    discovered_types.py       # Generated output schemas
  fetch/
    __init__.py
    fetch.py
    discovered_types.py

3.工作原理

当你要求AI代理处理你的数据时：

  1. 代理探索 可用的MCP工具 ./servers/
  2. 代理编写脚本 使用 call_mcp_tool() 从MCP服务器获取数据
  3. 脚本返回数据 -根据用例，可以是原始的或预处理的
  4. 代理处理结果 -可以总结、重塑或用作后续工具调用的输入

关键见解：并非所有处理都需要在脚本中进行。LLM可以处理后续交互中的摘要和数据转换。脚本侧重于高效的数据检索。

代理可能编写的示例脚本：

"""Analyze git repository commits."""

import asyncio
from runtime.mcp_client import call_mcp_tool

async def main():
    # Fetch recent commits
    result = await call_mcp_tool(
        "git__git_log",
        {"repo_path": ".", "max_count": 10}
    )

    # Return data for agent to process
    # Agent can then summarize, analyze, or use as input to other tools
    print(f"Fetched commit log")
    return result

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

4.代理执行

当代理需要运行脚本时，它会使用以下工具：

# The agent runs this automatically
uv run python -m runtime.harness workspace/my_script.py

# Or via the convenience alias
uv run mcp-exec workspace/my_script.py

该工具管理MCP客户端生命周期，连接到服务器，并捕获脚本的输出以发送回代理。

建筑

渐进式披露模式

传统方法 （高令牌使用率）：

Agent → MCP Server → [Full Data 50KB] → Agent processes all

渐进呈现 （减少98.7%）：

Agent → Discovers tools → Writes script
Script → MCP Server → [Full Data 50KB] → Returns to Agent
Agent → Processes/summarizes → Uses in follow-up calls

关键组件

  • runtime/mcp_client.py：延迟加载MCP客户端管理器
  • runtime/harness.py：脚本执行环境
  • runtime/generate_wrappers.py：自动生成类型化包装
  • runtime/normalize_fields.py：处理API现场套管
  • runtime/schema_utils.py：JSON架构→ Pydantic转化

发展

运行测试

# All tests
uv run pytest

# Unit tests only
uv run pytest tests/unit/

# Integration tests
uv run pytest tests/integration/

# With coverage
uv run pytest --cov=src/runtime

代码质量

# Type checking
uv run mypy src/

# Formatting
uv run black src/ tests/

# Linting
uv run ruff check src/ tests/

# Format check
uv run black --check src/ tests/

项目脚本

# Generate wrappers from tool definitions
uv run mcp-generate

# (Optional) Generate discovery config with LLM parameter generation
uv run mcp-generate-discovery

# (Optional) Execute safe tools and infer schemas
uv run mcp-discover

# Execute a Python script with MCP tools available
uv run mcp-exec workspace/script.py

Python特定功能

Pydantic的类型安全

所有工具参数和结果均已Pydantic验证：

from runtime.servers.git import git_status, GitStatusParams

# Type-safe parameters
params = GitStatusParams(repo_path=".")

# Type-safe result
result = await git_status(params)

异步/等待支持

建立在 asyncio 为了实现高效的I/O：

import asyncio
from runtime.mcp_client import call_mcp_tool

async def main():
    # Concurrent tool calls
    results = await asyncio.gather(
        call_mcp_tool("git__git_status", {"repo_path": "."}),
        call_mcp_tool("git__git_log", {"repo_path": ".", "max_count": 5}),
    )

asyncio.run(main())

现场标准化

自动处理不一致的API套管：

from runtime.normalize_fields import normalize_field_names

# ADO returns lowercase, expects PascalCase
ado_response = {"system.title": "Task", "custom.priority": "High"}
normalized = normalize_field_names(ado_response, "ado")
# Result: {"System.title": "Task", "Custom.priority": "High"}

处理丢失的输出模式

许多MCP服务器不提供 outputSchema 在他们的工具定义中，这在MCP规范中是可选的。该项目使用LLM驱动的参数生成提供自动模式发现：

运作原理

  1. 生成发现配置 (mcp-generate-discovery)

    • 连接到所有已配置的MCP服务器
    • 使用Claude生成合理的测试参数 inputSchema
    • 根据模式将工具分为安全/危险/未知
    • 写入 discovery_config.json 供审核

  2. 查看和编辑 （手动步骤）

    • 查看生成的配置
    • 根据需要添加/删除工具
    • 必要时修改测试参数

  3. 发现模式 (mcp-discover)

    • 使用测试参数执行安全工具
    • 从实际反应推断出Pydantic模型
    • 写入 servers/{server}/discovered_types.py

工具分类

工具按安全性自动分类：

  • 安全：匹配模式的工具，如 get_*, list_*, read_*, fetch, search_*等等。
  • 危险：匹配模式的工具，如 delete_*, remove_*, update_*, write_*等等。
  • 未知：不符合任何模式的工具（需要手动审查）

默认情况下，危险工具被排除在自动发现之外。

示例

{
  "servers": {
    "github": {
      "safeTools": {
        "search_code": {"q": "language:python", "per_page": 1},
        "list_repositories": {"sort": "stars", "per_page": 1}
      }
    }
  },
  "metadata": {
    "generated": true,
    "generated_count": 2,
    "skipped_count": 3,
    "tools_skipped": {
      "dangerous": ["delete_repository"],
      "unknown": ["analyze_code", "deploy_release", "configure_webhook"]
    }
  }
}

例子

看 workspace/example_progressive_disclosure.py 以获取演示令牌缩减模式的完整示例。

贡献

# Install dev dependencies
uv sync --all-extras

# Run quality checks before committing
uv run black src/ tests/
uv run mypy src/
uv run ruff check src/ tests/
uv run pytest

许可证

麻省理工学院

参考文献

相关 MCP