ChatGoogleGenerativeAI调用MCP工具入门指南：从零搭建到生产环境避坑

1次阅读

没有评论

共计 3214 个字符，预计需要花费 9 分钟才能阅读完成。

MCP（Model Control Plane）是 Google 云平台提供的 AI 模型管理中间件，在生成式 AI 工作流中主要承担三大职能：

模型路由：根据请求特征自动选择最优模型版本（如根据 query 长度切换 base/large 模型）
流量控制：实现灰度发布、A/ B 测试等流量分配策略
性能优化：内置请求批处理、结果缓存等加速机制

相较于直接调用基础 API，MCP 工具能为 ChatGoogleGenerativeAI 带来 30% 以上的 TPS 提升，同时降低错误率。

90% 的首次调用失败源于认证问题，典型表现包括：

使用过期 Token（默认 1 小时失效）
未正确处理服务账号密钥轮换
项目权限配置遗漏 aiplatform.endpoints.predict 权限

生成式 AI 响应时间波动较大，需特别注意：

未设置分层超时（建议连接超时 3s，读取超时 60s）
重试策略不合理（推荐指数退避算法）
未区分冷启动 / 热节点差异

MCP 返回的嵌套 JSON 结构常导致解析错误：

# 错误示范
result = response['predictions'][0]  # 可能抛出 KeyError

# 正确做法
from google.protobuf import json_format
def parse_response(response):
    return json_format.MessageToDict(
        response._pb,
        preserving_proto_field_name=True
    )['predictions'][0]['content']

from google.auth.transport.requests import Request
from google.oauth2 import service_account

class TokenManager:
    def __init__(self, key_path):
        self.credentials = service_account.Credentials.from_service_account_file(
            key_path,
            scopes=["https://www.googleapis.com/auth/cloud-platform"]
        )

    def get_token(self):
        if not self.credentials.valid:
            self.credentials.refresh(Request())
        return self.credentials.token

# 使用示例
token_manager = TokenManager("service-account.json")
headers = {"Authorization": f"Bearer {token_manager.get_token()}",
    "Content-Type": "application/json"
}

import requests
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def call_mcp_api(payload):
    try:
        response = requests.post("https://us-central1-aiplatform.googleapis.com/v1/projects/{project}/locations/us-central1/publishers/google/models/generative-ai:predict",
            json=payload,
            headers=headers,
            timeout=(3, 60)
        )
        response.raise_for_status()
        return parse_response(response.json())
    except requests.exceptions.RequestException as e:
        log_error(f"API call failed: {str(e)}")
        raise

graph LR
    A[客户端] -->|gRPC| B(MCP 代理层)
    B --> C{模型路由决策}
    C -->| 低延迟 | D[Base 模型]
    C -->| 高精度 | E[Large 模型]
    D --> F[结果聚合]
    E --> F
    F --> G[缓存层]
    G --> A

adapter = requests.adapters.HTTPAdapter(
    pool_connections=20,
    pool_maxsize=100,
    max_retries=3
)
session = requests.Session()
session.mount("https://", adapter)

# 单条请求格式
{
  "instances": [{"prompt": "解释量子计算原理"}
  ]
}

# 批量请求优化
{
  "instances": [{"prompt": "问题 1"},
    {"prompt": "问题 2"},
    {"prompt": "问题 3"}
  ]
}

import redis
from hashlib import md5

r = redis.Redis(host='localhost', port=6379)

def cached_call(prompt):
    key = md5(prompt.encode()).hexdigest()
    cached = r.get(key)
    if cached:
        return cached.decode()

    result = call_mcp_api({"prompt": prompt})
    r.setex(key, 3600, result)  # 1 小时过期
    return result

开发环境：使用临时凭据（如 Vault 动态生成）
生产环境：Workload Identity 联合认证
禁止将密钥硬编码在代码中

from pydantic import BaseModel, constr

class PromptRequest(BaseModel):
    prompt: constr(max_length=1000)
    temperature: float = Field(0.7, ge=0, le=1)
    max_tokens: int = Field(100, ge=1, le=2048)

def sanitize_output(text):
    patterns = [r"\d{4}-\d{4}-\d{4}-\d{4}",  # 信用卡号
        r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b"  # 邮箱
    ]
    for pattern in patterns:
        text = re.sub(pattern, "[REDACTED]", text)
    return text