Adk MCP Code Execution Anthropic

使用Google ADK使用MCP执行代码

使用Google ADK演示Anthropic的“使用MCP的代码执行”模式的全面实现。实现 代币减少89.8% 通过将MCP工具转换为基于代码的API，实现重复操作。

🎯 这表明了什么

我们不直接调用MCP工具（在上下文中加载所有模式并通过模型传递结果），而是：

1. 代理编写Python代码 与MCP服务器进行内联HTTP调用
2. 代码在优化的Docker沙盒中执行 （比基线快79%）
3. 中间结果保留在代码中 -永远不要输入LLM上下文
4. 仅返回最终摘要 给代理人

关键发现： 传统方法中上下文增长O（1）常数vs O（n²）二次型

📊 测量结果

单一操作

• 传统：411个代币，2.5秒⚡ 赢家 （越简单越好）
• 代码执行：658个令牌，5秒

10顺序操作

• 传统： 已处理7268个令牌，26秒
• 代码执行： 已处理744个令牌，9秒⚡
• 节省：89.8%的代币，66%的时间

看 tests/TEST_RESULTS.md 用于完整分析

🏗️ 建筑

实施了两种方法

1.传统MCP代理 （通过McpToolset调用函数）

User Query → LLM (with all tool schemas) → Function Call
                                              ↓
                                         McpToolset → HTTP → MCP Server
                                              ↓
                                         Result → LLM Context (grows!)

2.代码执行代理 （此存储库的重点）

User Query → LLM → Generates Python Code
                        ↓
                   DockerCodeExecutorOptimized
                        ↓
                   Code makes HTTP calls → MCP Server
                        ↓
                   Process ALL results in code
                        ↓
                   Return ONLY summary → LLM

主要区别：传统方法在每次工具调用时重新处理上下文（二次成本），代码执行保持上下文不变。

📋 先决条件

• Python 3.11+
• 带有API密钥的Google Cloud帐户
• 正在运行MCP服务器（通过ngrok公开）
• UV包管理器（或pip）

🚀 快速开始

1.安装依赖项

# Using UV (recommended)
uv venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate
uv pip install -e ".[dev]"

# Or using pip
python -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"

2.设置环境

创建 .env 文件：

MCP_SERVER_URL=https://your-ngrok-url.ngrok.io
GOOGLE_API_KEY=your_google_api_key_here

3.运行演示

cd mcp_adk_prototype
python src/demo.py

🧪 运行测试

快速比较（最重要！）

# Demonstrates the 89.8% token reduction
python tests/test_repeated_operations.py

所有测试

# Individual agents
python tests/test_programmatic.py              # Code execution agent
python tests/test_traditional_programmatic.py  # Traditional MCP agent

# Performance comparisons
python tests/test_token_comparison_fixed.py    # Single operation comparison
python tests/test_repeated_operations.py        # Sequential operations (⭐ key test)

# Run all tests
pytest tests/ -v

测试结果

综合分析见：

• tests/TEST_RESULTS.md -完整的性能分析
• 测试/README.md -测试套件文档

主要发现：

• 交叉点：3-5次操作（以下为传统操作，以上为代码执行操作）
• 上下文增长：传统O（n²），代码执行O（1）
• 使用传统时：简单查询，\<20个工具，延迟关键
• 在以下情况下使用代码执行：复杂的工作流程，100多种工具，成本关键

⚡ 性能优化

DockerCodeExecutor优化（当前实现）

问题：内置代码执行器没有网络访问权限（无法访问MCP服务器）

解决方案：带有预构建映像的自定义Docker执行器

结果:

• 基线：每次执行1.65秒（每次运行时安装pip）
• 优化：每次执行0.34秒⚡
• 改进: 速度提高79%

# Dockerfile.executor-optimized
FROM python:3.11-slim
RUN pip install --no-cache-dir urllib3  # Pre-installed!
RUN useradd -m -u 1000 sandbox
USER sandbox
WORKDIR /workspace

构建：

docker build -f Dockerfile.executor-optimized -t mcp-executor-optimized:latest .

🔒 安全考虑

当前（DockerCodeExecutor优化）

• ✅ Docker容器隔离
• ✅ 网络仅限于MCP服务器URL
• ✅ 非root用户执行
• ✅ 30秒超时
• ⚠️ 适用于开发/测试

生产（GkeCodeExecutor）

对于生产，请使用gVisor升级到GKE：

from google.adk.code_executors import GkeCodeExecutor

executor = GkeCodeExecutor(
    namespace="agent-sandbox",
    timeout_seconds=600,
    cpu_limit="1000m",
    mem_limit="2Gi"
)

其他好处：

• 🔒 内核级隔离（gVisor）
• 🔒 系统调用过滤
• 🔒 资源限制
• 🔒 带有自动清理功能的短命舱

📚 关键概念

传统工具调用（McpToolset）

# Each operation is a separate LLM invocation
# Context grows with EVERY call

Turn 1: User: "Get customer 1"
  → LLM: function_call(get_customer, {id: 1})
  → Result in context: 315 tokens

Turn 2: User: "Get customer 2"
  → Previous context STILL THERE: 315 tokens
  → LLM: function_call(get_customer, {id: 2})
  → Result in context: 407 tokens (+92)

... continues growing ...

Turn 10: Context = 1,140 tokens
Total processed: 7,268 tokens (quadratic!)

使用MCP执行代码（此实现）

# Agent writes code ONCE with inline HTTP calls
# ALL operations happen in the code

code = """
import urllib.request
import json

MCP_URL = "https://your-ngrok-url/mcp"

def call_mcp(tool_name, arguments):
    # ... HTTP helper ...
    pass

# Process ALL 10 customers in code
for customer_id in range(1, 11):
    response = call_mcp("get_customer", {"customer_id": customer_id})
    # Process in code, not in LLM context!

print("Active: 8, Disabled: 2")  # Only this returns
"""

# Result: 744 tokens total (constant!)

关键洞察：中间结果（所有10条客户记录）永远不要进入LLM上下文！

🎓 学习成果

运行此实现后，您将了解：

1. 为什么代码执行不是普遍更好 -简单查询的传统优势
2. 二次成本问题 -传统的O（n²）处理与代码执行中的O（1）处理
3. 交叉点 -3-5个操作，其中代码执行变得有利
4. 优化Docker执行器 -预构建图像的速度提高了79%
5. MCP集成的两种方法:

- 传统：带函数调用的McpToolset - 代码执行：在生成的Python代码中内联HTTP调用

1. 何时使用每种方法 -基于操作计数、工具计数和延迟要求

🔧 扩展原型

添加更多MCP服务器

1. 更新 tool_generator.py 处理多个服务器
2. 生成单独的模块（例如。， mcp_tools/salesforce.py)
3. 用新功能更新代理指令

部署到GKE

1. 使用gVisor设置GKE集群
2. 创建具有RBAC权限的服务帐户
3. 切换到 GkeCodeExecutor
4. 配置资源限制

添加监控

import google.cloud.logging

# Log all code executions
# Track token usage
# Monitor execution times
# Alert on errors

📖 参考文献

• Anthropic：使用MCP执行代码
• 谷歌ADK文档
• 模型上下文协议

🤝 贡献

这是一个研究原型。欢迎扩展和改进！

📝 许可证

麻省理工学院许可证-可自由用于研究和教育。