MCP Server and Google ADK Multi-Tool System

MCP Server 和 Google ADK 多工具系统

系统概述

该系统由两个主要组件组成：

1. MCP Server：一个 TypeScript/Express 服务，提供文件操作、API 调用、会话数据管理和天气信息工具。
2. Google ADK Agent：一个 Python Agent，通过 webhook 接口连接到 MCP Server 并使用其工具。

通信流程如下：

• Google ADK Agent 接收用户请求
• Agent 确定要使用的工具
• Agent 向 MCP Server 的 webhook 端点发送请求
• MCP Server 处理请求并执行操作
• MCP Server 将结果返回给 Agent
• Agent 将响应整理后发送给用户

运行所有组件

本指南将帮助你设置并运行 MCP Server 和 Google ADK Agent 系统，包括向系统添加新工具。

先决条件

• Node.js（v16+），用于 MCP Server
• Python（v3.9+），用于 Google ADK Agent
• 已安装 Google ADK SDK（pip install google-adk）
• 已启用 Vertex AI 的 Google 项目
• Google ADK Agent 访问 Vertex AI 所需的 IAM 权限
• MCP Server 用于访问代码仓库的 GitHub 或 Gitlab 访问令牌（作为环境变量）

设置环境

1. 设置 MCP serve 和 Google ADK Agentr：

   # run this from the root directory of the project to setup the MCP server and the Google ADK agent
   ./scripts/setup.sh
   gcloud auth application-default login
   

2. 配置环境变量：

   • 对于 MCP Server，请确保 mcp/ 中的 .env 包含：

     # Server Configuration
      PORT=9000
      BASE_DIR=./data
     
      # Repository Access Tokens, allows the MCP server to access private repositories
      GITHUB_ACCESS_TOKEN=your_github_token_here
      GITLAB_ACCESS_TOKEN=your_gitlab_token_here
     
      # Optional Configuration
      REPO_DIR=./repos
      MAX_EVENT_LISTENERS=100
     

   • 对于 Google ADK Agent，请确保 mcp_agent/ 中的 .env 包含：

     GOOGLE_CLOUD_PROJECT="your-google-project-id"
     GOOGLE_CLOUD_LOCATION="us-central1"
     GOOGLE_GENAI_USE_VERTEXAI="True"
     MCP_SERVER_URL=http://localhost:9000
     

3. 运行 MCP Server：

   # run this from the root directory of the project to start both the MCP server and the Google ADK agent
    ./scripts/run-mcp.sh
   

4. 运行 Google ADK Agent：

    # run this from the root directory of this repository to start the Google ADK agent
    
    # will start this in web mode
    ./scripts/run-agent.sh web
    
    # will start this in terminal mode
    ./scripts/run-agent.sh run
   

5. 打开 Web 界面并选择 mcp_agent：

   • 在浏览器中访问 http://localhost:8000
   • 从下拉菜单中选择 mcp_agent
   • 与 Agent 聊天，确认其正常运行

6. 测试现有工具：

在 Kubernetes 中运行 Google ADK Agent

要在 Kubernetes 中运行 Google ADK Agent，请按以下步骤操作：

1. 构建 Docker 镜像：

   docker build -t mcp-agent:latest -f mcp_agent/Dockerfile .
   

2. 将镜像推送到 Google Artifact Registry：

   # Build and tag the image
   

gcloud builds submit
--tag $GOOGLE_CLOUD_LOCATION-docker.pkg.dev/$GOOGLE_CLOUD_PROJECT/mcp-agent-repo/mcp-agent:latest
--project=$GOOGLE_CLOUD_PROJECT
.

Verify the image was pushed

gcloud artifacts docker images list
$GOOGLE_CLOUD_LOCATION-docker.pkg.dev/$GOOGLE_CLOUD_PROJECT/mcp-agent-repo
--project=$GOOGLE_CLOUD_PROJECT
--format=json

3. **创建 Kubernetes 部署**：
```yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: mcp-agent
spec:
replicas: 1
selector:
 matchLabels:
   app: mcp-agent
template:
 metadata:
   labels:
     app: mcp-agent
 spec:
   serviceAccount: mcp-agent-sa
   containers:
   - name: mcp-agent
     imagePullPolicy: Always
     image: us-central1-docker.pkg.dev/your-gar-registry/mcp-agent-repo/mcp-agent:latest
     resources:
       limits:
         memory: "512Mi"
         cpu: "500m"
         ephemeral-storage: "1Gi"
       requests:
         memory: "256Mi"
         cpu: "250m"
         ephemeral-storage: "512Mi"
     ports:
     - containerPort: 8000
     env:
       - name: PORT
         value: "8000"
       - name: GOOGLE_CLOUD_PROJECT
         value: "your-google-project-id"
       - name: GOOGLE_CLOUD_LOCATION
         value: "us-central1"
       - name: GOOGLE_GENAI_USE_VERTEXAI
         value: "True"
       # Use a service name or an external access method for the MCP server
       - name: MCP_SERVER_URL
         value: "http://mcp-server-service:9000"
     volumeMounts:
     - name: credentials
       mountPath: "/app/credentials"
       readOnly: true
   volumes:
   - name: credentials
     secret:
       secretName: mcp-agent-credentials
---
apiVersion: v1
kind: Service
metadata:
name: mcp-agent
spec:       
type: LoadBalancer
ports:
 - port: 80
   targetPort: 8000
selector:
 app: mcp-agent

向 MCP Server 和 Google ADK Agent 系统添加新工具

本分步指南将介绍向 MCP Server 添加新工具并使其可供 Google ADK Agent 使用的完整流程。以下指南将以图像生成工具为例。

3. 向 MCP Server 添加新工具

我们来向 MCP Server 添加图像生成工具。接下来将逐步介绍所有必要的更改。

步骤 1：在 MCP Server 中创建工具实现

首先，我们来将核心图像生成功能添加到 MCP Server。在 mcp/app.ts 中的 app.ts 文件里添加以下内容：

// 1. Add new type for ImageGeneration operation
type ImageGenerationOptions = {
  prompt: string;
  width?: number; 
  height?: number;
  style?: string;
  format?: 'png' | 'jpeg' | 'webp';
  negativePrompt?: string;
};

// 2. Add the image generation tool implementation
const imageGenerationTool = async (options: ImageGenerationOptions): Promise<{success: boolean; data?: any; error?: string}> => {
  try {
    // Validate parameters
    if (!options.prompt) {
      return { success: false, error: 'Image prompt is required' };
    }
    
    // Set defaults for missing options
    const width = options.width || 512;
    const height = options.height || 512;
    const format = options.format || 'png';
    const style = options.style || 'photorealistic';
    
    console.log(`Generating image for prompt: "${options.prompt}" with style: ${style}, dimensions: ${width}x${height}`);
    
    // For this example, we'll just mock the image generation
    // In a real implementation, you would call an API like Stable Diffusion or DALL-E
    const mockImageData = {
      prompt: options.prompt,
      imageUrl: `https://example.com/generated_images/${Date.now()}.${format}`,
      width,
      height,
      style,
      format,
      generatedAt: new Date().toISOString()
    };
    
    // In a real implementation, you might store the image file
    // For mock purposes, write a metadata file
    const metadataPath = validatePath(`images/metadata_${Date.now()}.json`);
    const dir = path.dirname(metadataPath);
    await fs.mkdir(dir, { recursive: true });
    await fs.writeFile(metadataPath, JSON.stringify(mockImageData, null, 2), 'utf-8');
    
    return {
      success: true,
      data: mockImageData
    };
  } catch (error: any) {
    console.error('Image generation error:', error);
    return {
      success: false, 
      error: error.message || 'Unknown error during image generation'
    };
  }
};

步骤 2：为 Google ADK Webhook 添加处理函数

将此处理函数添加到 MCP Server，以处理来自 Google ADK Agent 的请求：

// Add this to the handler functions section
async function handleImageGenerationTool(parameters: any, sessionId: string) {
  // Validate required parameters
  if (!parameters.prompt) {
    return { 
      success: false, 
      error: 'Missing required parameter: prompt' 
    };
  }
  
  // Create options object for the image generation tool
  const options: ImageGenerationOptions = {
    prompt: parameters.prompt,
    width: parameters.width || 512,
    height: parameters.height || 512,
    style: parameters.style || 'photorealistic',
    format: parameters.format || 'png',
    negativePrompt: parameters.negativePrompt
  };
  
  return await imageGenerationTool(options);
}

步骤 3：更新 ADK Webhook 中的 Switch 语句

现在，在 /api/adk-webhook 路由处理程序的 switch 语句中添加一个新 case：

// Find the switch statement in app.post('/api/adk-webhook', ...) 
switch (toolName) {
  case 'file_system':
    result = await handleFileSystemTool(parameters, mcpSessionId);
    break;
  case 'api_call':
    result = await handleApiTool(parameters, mcpSessionId);
    break;
  case 'session_data':
    result = await handleSessionDataTool(parameters, mcpSessionId);
    break;
  case 'weather':
    result = await handleWeatherTool(parameters, mcpSessionId);
    break;
  // Add the new case for image generation
  case 'image_generation':
    result = await handleImageGenerationTool(parameters, mcpSessionId);
    break;
  default:
    result = {
      success: false,
      error: `Unknown tool name: ${toolName}`
    };
}

步骤 4：添加直接访问端点

添加一个专用端点，以便直接访问图像生成工具：

// Add this route to the app
app.post('/api/session/:sessionId/image', async (req, res) => {
  const { sessionId } = req.params;
  const session = getSessionAndUpdate(sessionId);

  if (!session) {
    return res.status(404).json({
      success: false,
      error: 'Session not found'
    });
  }

  const options: ImageGenerationOptions = req.body;

  if (!options.prompt) {
    return res.status(400).json({
      success: false,
      error: 'Image prompt is required'
    });
  }

  const result = await imageGenerationTool(options);

  if (result.success) {
    // Emit an event for SSE clients
    emitEvent(sessionId, 'image-generation', {
      imageUrl: result.data.imageUrl,
      prompt: options.prompt,
      timestamp: new Date().toISOString()
    });
    
    res.status(200).json(result);
  } else {
    res.status(400).json(result);
  }
});

步骤 5：更新 API 文档

在 /api/help 端点的 API 文档中添加新工具：

// Find the endpoints array in the helpDocs object
endpoints: [
  // Add these new entries
  {
    path: "/api/session/{sessionId}/image",
    method: "POST",
    description: "Generate an image from a text prompt",
    parameters: [
      {
        name: "sessionId",
        in: "path",
        required: true,
        description: "Session identifier"
      }
    ],
    requestBodyExample: {
      prompt: "A beautiful sunset over mountains",
      width: 512,
      height: 512,
      style: "photorealistic",
      format: "png",
      negativePrompt: "blur, low quality"
    },
    responseExample: {
      success: true,
      data: {
        prompt: "A beautiful sunset over mountains",
        imageUrl: "https://example.com/generated_images/1717451623456.png",
        width: 512,
        height: 512,
        style: "photorealistic",
        format: "png",
        generatedAt: "2025-06-03T12:00:00.000Z"
      }
    },
    curlExample: `curl -X POST ${baseUrl}/api/session/{sessionId}/image \\
  -H "Content-Type: application/json" \\
  -d '{"prompt": "A beautiful sunset over mountains", "style": "photorealistic"}'`
  },
  // Add the Google ADK webhook documentation for image_generation tool
  {
    path: "/api/adk-webhook",
    method: "POST",
    description: "Webhook for Google ADK image generation",
    requestBodyExample: {
      session_id: "google-adk-session-123",
      tool_name: "image_generation",
      parameters: {
        prompt: "A beautiful sunset over mountains",
        width: 512,
        height: 512,
        style: "photorealistic"
      },
      request_id: "request-123"
    },
    responseExample: {
      success: true,
      data: {
        prompt: "A beautiful sunset over mountains",
        imageUrl: "https://example.com/generated_images/1717451623456.png",
        width: 512,
        height: 512,
        style: "photorealistic",
        format: "png",
        generatedAt: "2025-06-03T12:00:00.000Z"
      },
      mcp_session_id: "550e8400-e29b-41d4-a716-446655440000",
      request_id: "request-123"
    },
    notes: "This endpoint is used by the Google ADK agent to generate images."
  }
],

4. 为 Google ADK Agent 添加工具支持

现在，我们来将图像生成工具添加到 Google ADK Agent。该 Agent 位于 mcp_agent。

步骤 1：向 MCPToolkit 类添加方法

首先，在 mcp_agent/mcp_toolkit.py 文件中添加一个新方法，以调用图像生成工具：

# Add this method to the MCPToolkit class in mcp_toolkit.py
def generate_image(self, prompt: str, width: int = 512, height: int = 512, 
                  style: str = "photorealistic", format: str = "png", 
                  negative_prompt: str = None) -> Dict:
    """Generate an image from a text prompt using the MCP server"""
    params = {
        "prompt": prompt,
        "width": width,
        "height": height,
        "style": style,
        "format": format
    }
    
    if negative_prompt:
        params["negativePrompt"] = negative_prompt
        
    return self.execute_tool("image_generation", params)

步骤 2：向 tools.py 添加工具函数

在 tools.py 中创建一个新的工具函数，供 ADK Agent 使用：

# Add this to the tools.py file
def mcp_generate_image(prompt: str, style: str = "photorealistic", width: int = 512, height: int = 512) -> dict:
    """Generates an image from a text prompt.
    
    Args:
        prompt: Text description of the image to generate
        style: Style for the image (e.g., photorealistic, cartoon, sketch)
        width: Width of the output image in pixels
        height: Height of the output image in pixels
        
    Returns:
        dict: A dictionary with status ('success' or 'error') and either image info or error message
    """
    try:
        result = mcp_toolkit.generate_image(
            prompt=prompt, 
            style=style,
            width=width, 
            height=height
        )
        
        if result.get("success"):
            image_data = result.get("data", {})
            return {
                "status": "success",
                "image_url": image_data.get("imageUrl"),
                "message": f"Generated image for prompt: '{prompt}' in {style} style."
            }
        else:
            return {
                "status": "error",
                "error_message": result.get("error", "Unknown error generating image")
            }
    except Exception as e:
        logger.error(f"Error in mcp_generate_image: {str(e)}")
        return {
            "status": "error",
            "error_message": f"Exception: {str(e)}"
        }

步骤 3：向 Agent 注册工具

更新 agent.py 文件，将新工具添加进去：

# Add to the imports in agent.py
from .tools import (
    mcp_read_file, 
    mcp_write_file, 
    mcp_list_files, 
    mcp_delete_file,
    mcp_get_weather,
    mcp_call_api,
    mcp_store_data,
    mcp_store_number,
    mcp_store_boolean,
    mcp_retrieve_data,
    mcp_generate_image  # Add this import
)

# Update the agent definition
agent = Agent(
    name="mcp_agent",
    model="gemini-2.0-flash",
    description="Agent that can handle weather, time, and interact with a Model Control Protocol server",
    instruction="""I can help you with various tasks through my integration with the MCP server.
I can:
- Get current time in different cities
- Check weather conditions in locations
- Read, write, list, and delete files
- Make API calls to external services
- Store and retrieve data in a session (text, numbers, or boolean values)
- Generate images from text descriptions

When you ask me about files, I'll use the appropriate file operation tools.
When you ask me about weather, I'll look up the latest conditions.
When you want to store information for later, I'll use session storage.
When you ask me to create an image, I'll generate one based on your description.
""",
    tools=[
        # MCP file system tools
        mcp_read_file,
        mcp_write_file,
        mcp_list_files,
        mcp_delete_file,
        
        # MCP API tools
        mcp_call_api,
        mcp_get_weather,
        
        # MCP session tools
        mcp_store_data,
        mcp_store_number, 
        mcp_store_boolean, 
        mcp_retrieve_data,
        
        # New image generation tool
        mcp_generate_image  # Add this tool
    ]
)

5. 测试和调试

使用 cURL 直接测试工具

使用 cURL 直接测试图像生成端点：

# First create a session
export SESSION=$(curl -X POST http://localhost:8080/api/session | jq -r '.sessionId')

# Then call the image generation endpoint
curl -X POST http://localhost:8080/api/session/$SESSION/image \
  -H "Content-Type: application/json" \
  -d '{
    "prompt": "A beautiful sunset over mountains",
    "style": "photorealistic"
  }'

通过 ADK Webhook 测试

通过 ADK Webhook 测试图像生成：

curl -X POST http://localhost:8080/api/adk-webhook \
  -H "Content-Type: application/json" \
  -d '{
    "session_id": "test-session-123",
    "tool_name": "image_generation",
    "parameters": {
      "prompt": "A beautiful sunset over mountains",
      "style": "photorealistic"
    },
    "request_id": "test-request-1"
  }'

运行 Agent 并测试工具

运行 Agent，并使用提示词测试图像生成工具：

cd /mcp-server-google-adk-multi-tool-system
python -m mcp_agent.main

Agent 运行后，尝试：

You: Generate an image of a cat playing piano

6. 当前集成的工作原理

让我们来看看系统中现有集成的工作原理。

MCP 服务组件

1. 会话管理：

   • 每个客户端都会获得唯一的会话 ID
   • 会话会存储用户数据，并在 30 分钟后过期
   • 会话使用内存中的 Map 管理

2. 工具实现：

   • 每个工具（fileSystemTool、apiTool 等）都通过异步函数实现
   • 工具负责验证、处理和错误处理
   • 工具会返回标准化的响应对象，其中包含 success、data 和可选的 error 字段

3. 端点：

   • 每个工具都有专用端点（例如，/api/session/:sessionId/filesystem）
   • 中央 Webhook 端点（/api/adk-webhook）用于处理来自 Google ADK 的请求
   • Webhook 端点会根据 tool_name 将请求路由到相应的处理函数

4. 服务器发送事件（SSE）：

   • /api/sse/:sessionId 端点支持实时更新
   • emitEvent 函数会向已连接的客户端发送事件
   • 客户端可以通过 EventSource 连接监听事件

Google ADK Agent 组件

1. MCPToolkit 类：

   • 管理与 MCP 服务的通信
   • 处理会话的创建和管理
   • 为每个工具操作提供方法
   • 维护 SSE 连接，以支持实时更新

2. 工具函数：

   • 每个工具函数（mcp_read_file、mcp_get_weather 等）都是 MCPToolkit 方法的封装
   • 函数遵循 Google ADK 格式，并提供适当的文档和类型提示
   • 函数会返回标准化的响应对象，其中包含 status 和其他字段

3. Agent 定义：

   • Google ADK SDK 中的 Agent 类定义了 Agent 的能力
   • Agent 注册工具函数并提供指令
   • Agent 会处理用户输入的 NLU（自然语言理解）

4. 主运行器：

   • 初始化 Agent 和工具包
   • 设置事件监听器
   • 管理对话循环
   • 处理错误并执行清理

组件间的数据流

1. 用户请求流程：

   • 用户向 Agent 发送文本查询
   • Agent 处理查询并确定合适的工具
   • Agent 使用提取的参数调用工具函数
   • 工具函数调用 MCPToolkit 方法
   • MCPToolkit 向 MCP 服务 Webhook 发送请求
   • MCP 服务处理请求并返回结果
   • 结果沿相同链路反向返回

2. 服务器发送事件流程：

   • MCP 服务处理某项操作
   • 服务使用 emitEvent 发出事件
   • SSE 连接将事件传输给客户端
   • MCPToolkit 在 _sse_worker 线程中接收事件
   • 事件回调处理该事件

概述

向 MCP 服务和 Google ADK Agent 添加新工具涉及以下关键步骤：

1. MCP 服务：

   • 创建工具实现函数
   • 为 Webhook 添加处理函数
   • 更新 Webhook switch 语句
   • 根据需要添加专用端点
   • 更新 API 文档

2. Google ADK Agent：

   • 向 MCPToolkit 类添加方法
   • 在 tools.py 中创建工具函数
   • 向 Agent 注册工具
   • 更新 Agent 指令

按照本指南，你可以轻松扩展 MCP 服务和 Google ADK Agent，添加新的工具和功能。模块化设计让你可以轻松添加新功能，同时保持一致的接口和错误处理方式。

作者：Andrew Milam [email protected]