Google Adk SSE Multi Tool System

MCP服务器与Google ADK多工具系统

系统概述

该系统由两个主要部分组成：

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

通信流程为：

• Google ADK代理接收用户请求
• Agent决定使用哪个工具
• 代理向MCP服务器的webhook端点发送请求
• MCP服务器处理请求并执行操作
• MCP服务器将结果返回给代理
• 代理为用户格式化响应

运行一切

本指南将帮助您设置和运行MCP服务器和Google ADK代理系统，包括向系统添加新工具。

先决条件

• MCP服务器的Node.js（v16+）
• 适用于Google ADK代理的Python（v3.9+）
• 已安装Google ADK SDK(pip install google-adk)
• 启用Vertex AI的谷歌项目
• Google ADK代理访问Vertex AI的IAM权限
• GitHub或Gitlab访问令牌作为MCP服务器访问存储库的环境变量

设置环境

1. 设置MCP服务和谷歌ADK代理:

   # 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

1. 配置环境变量:

- 对于MCP服务器，请确保在 mcp/ 包含：

     # 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代理，请确保在 mcp代理/ 包含：

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

1. 运行MCP服务器:

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

1. 运行谷歌ADK代理:

    # 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

1. 打开web界面并选择mcp_agent:

- 首选 http://localhost:8000 浏览器中 - 选择 mcp_agent 从下拉列表中 - 与客服聊天以确保其正常工作

1. 测试现有工具:

在Kubernetes中运行Google ADK代理

要在Kubernetes中运行Google ADK代理，请执行以下步骤：

1. 构建Docker镜像:

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

1. 将图像推送到谷歌工件注册表:

   # Build and tag the image

gcloud构建提交 \ --标签$GOOGLE_CLOUD_LOCATION-docker.pkg.dev/$GOOGLE_CLOU项目/mcp代理仓库/mcp代理：最新 \ --项目=$GOOGLE_CLOUD_project \ .

验证图像是否已推送

gcloud工件docker镜像列表 \ $GOOGLE_CLOUD_LOCATION-docker.pkg/dev/$GOOGLE_CLOUD_PROJECT/mcp代理仓库 \ --项目=$GOOGLE_CLOUD_project \ --格式=json

3. **Create a Kubernetes deployment**:

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服务器和Google ADK代理系统中添加新工具

本分步指南将介绍向MCP服务器添加新工具并使其可供Google ADK代理使用的整个过程。下面的指南将使用图像生成工具作为我们的示例。

### 3.向MCP服务器添加新工具

让我们向MCP服务器添加一个图像生成工具。我们将逐步完成所有必要的更改。

#### 步骤1：在MCP服务器中创建工具实现

首先，让我们将核心映像生成功能添加到MCP服务器。将其添加到app.ts文件中 [mcp/app.ts](mcp/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 => { 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' }; } };

#### 第二步：为Google ADK Webhook添加处理函数

将此处理程序函数添加到MCP服务器，以处理来自Google ADK代理的请求：

// 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语句

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

// 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文档中 `/api/help` 端点：

// 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: "", 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: "", 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代理添加工具支持

现在，让我们将图像生成工具添加到Google ADK代理中，该代理位于 [mcp代理](mcp_agent).

#### 步骤1：向MCPToolkit类添加方法

首先，向 [mcp_agent/mcptoolkit.py](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](mcp_agent/tools.py) 其将暴露于ADK试剂：

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.py](mcp_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" }'

#### 运行代理并测试工具

运行代理并测试图像生成工具，并显示提示：

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

然后，当代理运行时，尝试：

You: Generate an image of a cat playing piano

### 6.当前集成如何工作

让我们来看看现有的集成在系统中是如何工作的。

#### MCP服务器组件

1. **会话管理**:

   - 每个客户端都有一个唯一的会话ID
   - 会话存储用户数据，有效期为30分钟
   - 会话使用内存中的映射进行管理

1. **工具实施**:

   - 每个工具（fileSystemTool、apiTool等）都实现为异步函数
   - 工具处理验证、处理和错误处理
   - 工具返回一个标准化的响应对象 `success`, `data`，可选 `error` 字段

1. **端点**:

   - 每个工具具有专用端点（例如。， `/api/session/:sessionId/filesystem`)
   - 中心webhook端点(`/api/adk-webhook`)处理来自Google ADK的请求
   - webhook端点根据以下内容将请求路由到适当的处理程序函数 `tool_name`

1. **服务器发送事件（SSE）**:

   - 这 `/api/sse/:sessionId` 端点支持实时更新
   - 这 `emitEvent` 函数将事件发送到连接的客户端
   - 客户端可以通过EventSource连接监听事件

### Google ADK代理组件

1. **MCPToolkit类**:

   - 管理与MCP服务器的通信
   - 处理会话创建和管理
   - 为每个工具操作提供方法
   - 保持SSE连接以进行实时更新

1. **工具功能**:

   - 每个工具函数（mcp_read_file、mcp_get_weather等）都是MCPToolkit方法的包装器
   - 函数遵循Google ADK格式，并带有适当的文档和类型提示
   - 函数返回标准化的响应对象 `status` 以及其他字段

1. **代理定义**:

   - Google ADK SDK中的Agent类定义了代理功能
   - Agent注册工具功能并提供说明
   - Agent处理用户输入的NLU（自然语言理解）

1. **主流道**:

   - 初始化代理和工具包
   - 设置事件侦听器
   - 管理对话循环
   - 处理错误和清理

### 组件之间的数据流

1. **用户请求流**:

   - 用户向代理发送文本查询
   - Agent处理查询并确定适当的工具
   - 代理使用提取的参数调用工具函数
   - 工具函数调用MCPToolkit方法
   - MCPToolkit向MCP服务器webhook发送请求
   - MCP服务器处理请求并返回结果
   - 结果反向通过同一链返回

1. **服务器发送的事件流**:

   - MCP服务器处理操作
   - 服务器发出事件 `emitEvent`
   - SSE连接将事件传输到客户端
   - MCPToolkit在中接收事件 `_sse_worker` 线程
   - 事件回调处理事件

## 总结

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

1. **MCP服务器**:

   - 创建工具实现功能
   - 为webhook添加处理函数
   - 更新webhook开关语句
   - 如果需要，添加专用端点
   - 更新API文档

1. **谷歌ADK代理**:

   - 向MCPToolkit类添加方法
   - 在tools.py中创建一个工具函数
   - 向代理注册工具
   - 更新代理说明

通过遵循本指南，您可以使用新的工具和功能轻松扩展MCP服务器和Google ADK代理。模块化设计使添加新功能变得简单，同时保持一致的界面和错误处理方法。

作者：安德鲁·米拉姆andrewmichaelmilam@gmail.com