【免费下载链接】python-docs-samples
Code samples used on cloud.google.com
项目地址:
https://gitcode.com/GitHub_Trending/py/python-docs-samples
点击查看 免费下载
本文基于 python-docs-samples 仓库中的 run/mcp-server 示例,系统讲解如何将一个基于 FastMCP 的远程 MCP(Model Context Protocol)服务器部署到 Cloud Run,并通过 streamable-http 传输、IAM 认证、Cloud Run Proxy 隧道与 OpenTelemetry 观测完成从开发到上线的全流程。读完本文,你将掌握远程 MCP 服务器的部署命令、客户端认证机制、本地代理测试方法以及可观测性配置,能够直接复现一个可被团队成员共享访问的云端 MCP 服务。
为什么要把 MCP 服务器运行在云端
MCP(Model Context Protocol)是让 LLM 应用与外部工具、数据源交互的开放协议。传统上,MCP 服务器多运行在开发者本机,但正如 run/mcp-server/README.md 所述,把 MCP 服务器远程部署到 Cloud Run 可以获得三个核心收益:
- 弹性伸缩(Scalability):Cloud Run 基于请求量自动扩缩容,MCP 服务器无需预留资源即可平滑应对突发调用,空闲时缩到零,不产生多余费用。
- 集中式服务器(Centralized server):团队成员不再需要各自在本机维护一套 MCP 服务器,而是通过 IAM 权限共享同一个集中式服务。当服务器更新了新工具或修复了缺陷,所有成员立即受益,避免了"各跑各的、版本漂移"的维护成本。
- 安全可控(Security):Cloud Run 原生支持强制认证请求,可以只允许经过授权的连接访问 MCP 服务器,防止未授权访问和滥用。
[!IMPORTANT] 安全这一点至关重要。如果部署时不强制认证,公网上的任何人都可能访问并调用你的 MCP 服务器,进而触发其中的工具造成系统损害。因此本示例的所有部署命令都显式携带了 –no-allow-unauthenticated 参数。
示例概览:一个基于 FastMCP 的数学 MCP 服务器
为什么选择数学工具作为示例
LLM 在非确定性任务上表现出色——理解意图、生成创意文本、总结复杂观点、对抽象概念推理;但在确定性任务上却极不可靠——那些只有一个且唯一正确答案的计算。加减法就是典型代表。
本示例通过把 add、subtract 两个数学工具以 MCP 工具的形式暴露给 LLM,演示了"用确定性工具补偿 LLM 短板"这一 MCP 的核心价值场景:LLM 不再自己"猜"计算结果,而是调用经过验证的工具获得精确答案。
核心实现:server.py
示例使用 FastMCP:
# [START cloudrun_mcpserver]
import asyncio
import logging
import os
from fastmcp import FastMCP
logger = logging.getLogger(__name__)
logging.basicConfig(format="[%(levelname)s]: %(message)s", level=logging.INFO)
mcp = FastMCP("MCP Server on Cloud Run")
@mcp.tool()
def add(a: int, b: int) -> int:
"""Use this to add two numbers together.
Args:
a: The first number.
b: The second number.
Returns:
The sum of the two numbers.
"""
logger.info(f">>> 🛠️ Tool: 'add' called with numbers '{a}' and '{b}'")
return a + b
@mcp.tool()
def subtract(a: int, b: int) -> int:
"""Use this to subtract two numbers.
Args:
a: The first number.
b: The second number.
Returns:
The difference of the two numbers.
"""
logger.info(f">>> 🛠️ Tool: 'subtract' called with numbers '{a}' and '{b}'")
return a – b
if __name__ == "__main__":
logger.info(f"🚀 MCP server started on port {os.getenv('PORT', 8080)}")
# Could also use 'sse' transport, host="0.0.0.0" required for Cloud Run.
asyncio.run(
mcp.run_async(
transport="streamable-http",
host="0.0.0.0",
port=os.getenv("PORT", 8080),
)
)
# [END cloudrun_mcpserver]
从源码(server.py)可以看到几个关键点:
- 通过 @mcp.tool() 装饰器将普通 Python 函数(add、subtract)注册为 MCP 工具,函数的类型注解(a: int, b: int -> int)与 docstring 会被 FastMCP 自动转换为工具元数据,供 LLM 理解调用方式。
- 每个工具内部记录结构化日志,方便在 Cloud Logging 中追踪每次调用及其入参。
- 入口处使用 transport="streamable-http",这是让 MCP 服务器可远程运行的关键传输层选择;代码注释同时指出也可改用 sse 传输。无论哪种传输,绑定 host="0.0.0.0" 都是 Cloud Run 容器环境所必需的,否则服务无法对外接收流量。
- 端口通过环境变量 PORT 读取,默认 8080,与 Cloud Run 的容器端口约定保持一致。
依赖清单:pyproject.toml
项目的依赖由 pyproject.toml 声明,并使用 uv 管理:
[project]
name = "mcp-server"
version = "0.1.0"
description = "Example of deploying an MCP server on Cloud Run"
readme = "README.md"
requires-python = ">=3.10"
dependencies = [
"fastmcp==3.4.3",
"opentelemetry-api==1.40.0",
"opentelemetry-sdk==1.40.0",
"opentelemetry-exporter-otlp-proto-grpc==1.40.0",
"google-auth==2.49.1",
"grpcio==1.80.0",
]
其中 requires-python = ">=3.10" 对应 README 中 Python 3.10+ 的前置要求;opentelemetry-* 与 google-auth、grpcio 为可观测性模块提供支撑(详见后文"OpenTelemetry 观测"一节)。依赖版本的精确锁定记录在 uv.lock 中,保证可复现构建。
环境准备与项目设置
前置条件
根据 run/mcp-server/README.md,部署前需要准备:
- Python 3.10+:满足 pyproject.toml 中 requires-python = ">=3.10" 的要求;
- uv:用于包管理与项目运行(uv run、uv sync 等命令均依赖它);
- Google Cloud SDK(gcloud):用于认证、部署与代理。
设置认证与项目
gcloud auth login
export PROJECT_ID=<your-project-id>
gcloud config set project $PROJECT_ID
gcloud auth login 完成用户身份认证;export PROJECT_ID 设置环境变量;gcloud config set project 将当前项目切换为目标 GCP 项目。后续的部署、建仓、代理命令都会使用该配置。
将 MCP 服务器部署到 Cloud Run
README 提供了两种部署方式:从源码直接部署与通过容器镜像部署。两条路径都以 –no-allow-unauthenticated 强制要求认证——这是出于安全考虑的必要设置:若不禁用匿名访问,任何人都能调用你的 MCP 服务器,可能对系统造成损害。
方式一:从源码直接部署
gcloud run deploy mcp-server –no-allow-unauthenticated –region=us-central1 –source .
–source . 让 Cloud Run 基于当前目录源码构建并部署,同时自动识别 Dockerfile。若本地尚未安装 Docker,Cloud Run 也会提示并引导使用 Cloud Build 完成构建。
方式二:通过容器镜像部署
第一步,创建 Artifact Registry 仓库,用于存放容器镜像:
gcloud artifacts repositories create mcp-servers \\
–repository-format=docker \\
–location=us-central1 \\
–description="Repository for remote MCP servers" \\
–project=$PROJECT_ID
第二步,用 Cloud Build 构建并推送镜像:
gcloud builds submit –region=us-central1 –tag us-central1-docker.pkg.dev/$PROJECT_ID/mcp-servers/mcp-server:latest
第三步,将镜像部署到 Cloud Run:
gcloud run deploy mcp-server \\
–image us-central1-docker.pkg.dev/$PROJECT_ID/mcp-servers/mcp-server:latest \\
–region=us-central1 \\
–no-allow-unauthenticated
部署成功的标志
服务部署成功后,gcloud 会输出类似信息:
Service [mcp-server] revision [mcp-server-12345-abc] has been deployed and is serving 100 percent of traffic.
此时流量已 100% 由新修订版本承载,MCP 服务器对外可用。
Dockerfile 解析
两种部署方式最终都会使用 Dockerfile 构建容器,其关键步骤值得细读:
# Use the official Python image
FROM python:3.14-slim
# Install uv
COPY –from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/
# Install the project into /app
COPY . /app
WORKDIR /app
# Allow statements and log messages to immediately appear in the logs
ENV PYTHONUNBUFFERED=1
# Install dependencies
RUN uv sync
EXPOSE $PORT
# Run the FastMCP server
CMD ["uv", "run", "server.py"]
- 基础镜像选用 python:3.14-slim,保持轻量;
- 通过多阶段 COPY –from=ghcr.io/astral-sh/uv:latest 直接引入 uv 二进制,无需额外安装步骤;
- ENV PYTHONUNBUFFERED=1 让日志立即输出,确保 Cloud Run 控制台能实时捕获服务器日志;
- RUN uv sync 依据 pyproject.toml/uv.lock 安装精确版本的依赖;
- EXPOSE $PORT 与 CMD ["uv", "run", "server.py"] 配合,容器启动即运行 FastMCP 服务器。
认证 MCP 客户端
由于部署时指定了 –no-allow-unauthenticated,任何连接远程 MCP 服务器的客户端都必须完成认证。Cloud Run 官方文档"Host MCP servers on Cloud Run"一节对此有更全面的说明,具体认证方式取决于 MCP 客户端运行的位置。
IAM 角色:roles/run.invoker
默认情况下,Cloud Run 服务的 URL 要求所有请求都带有 Cloud Run Invoker(roles/run.invoker)IAM 角色的授权。这一策略绑定提供了强安全机制,确保你的本地 MCP 客户端是通过合法身份访问服务。
因此,你以及任何试图访问远程 MCP 服务器的团队成员,都必须在其 Google Cloud 账号上绑定 roles/run.invoker 角色,否则请求会被拒绝。
使用 Cloud Run Proxy 建立认证隧道
对于本示例,最直接的认证方式是运行 Cloud Run proxy,在本地机器与远程 MCP 服务器之间建立一个已认证的加密隧道:
gcloud run services proxy mcp-server –region=us-central1
[!TIP] 如果本地尚未安装 Cloud Run proxy,上述命令可能会提示下载,按提示完成下载与安装即可。
运行成功后输出如下:
Proxying to Cloud Run service [mcp-server] in project [<YOUR_PROJECT_ID>] region [us-central1]
http://127.0.0.1:8080 proxies to https://mcp-server-abcdefgh-uc.a.run.app
此后,所有发往 http://127.0.0.1:8080 的流量都会被自动认证并转发到远程 MCP 服务器,本地客户端无需自行处理令牌签发与 IAM 签名。
测试远程 MCP 服务器
测试客户端源码解析
仓库提供了专门的测试脚本 test_server.py,它使用 FastMCP 客户端连接本地代理端口并验证工具调用:
# [START cloudrun_mcpserver_test]
import asyncio
from fastmcp import Client
async def test_server():
# Test the MCP server using streamable-http transport.
# Use "/sse" endpoint if using sse transport.
async with Client("http://localhost:8080/mcp") as client:
# List available tools
tools = await client.list_tools()
for tool in tools:
print(f">>> 🛠️ Tool found: {tool.name}")
# Call add tool
print(">>> 🪛 Calling add tool for 1 + 2")
result = await client.call_tool("add", {"a": 1, "b": 2})
print(f"<<< ✅ Result: {result.content[0].text}")
# Call subtract tool
print(">>> 🪛 Calling subtract tool for 10 – 3")
result = await client.call_tool("subtract", {"a": 10, "b": 3})
print(f"<<< ✅ Result: {result.content[0].text}")
if __name__ == "__main__":
asyncio.run(test_server())
# [END cloudrun_mcpserver_test]
从源码(test_server.py)可以看到完整调用链:
- 客户端连接地址为 http://localhost:8080/mcp——注意末尾的 /mcp 路径,这是 streamable-http 传输的端点约定;代码注释说明若改用 sse 传输,则应连接 /sse 端点。
- client.list_tools() 列出服务器公开的全部工具,用于校验工具注册是否正确。
- client.call_tool("add", {"a": 1, "b": 2}) 以 JSON 参数形式调用工具,返回结果通过 result.content[0].text 读取文本内容。
运行测试
[!NOTE] 运行测试前,请确保 Cloud Run proxy 正在运行(即上一步的 gcloud run services proxy 命令没有退出)。
在新的终端中运行:
uv run test_server.py
预期输出如下:
>>> 🛠️ Tool found: add
>>> 🛠️ Tool found: subtract
>>> 🪛 Calling add tool for 1 + 2
<<< ✅ Result: 3
>>> 🪛 Calling subtract tool for 10 – 3
<<< ✅ Result: 7
add(1, 2) 返回 3,subtract(10, 3) 返回 7,说明远程 MCP 服务器已经成功部署并通过 FastMCP 客户端完成端到端验证。
使用 OpenTelemetry 进行可观测性监控
原理:FastMCP 原生支持 OpenTelemetry
本示例集成了 OpenTelemetry,将 traces、logs、metrics 发送到 Google Cloud Observability(Cloud Trace、Cloud Logging、Cloud Monitoring)。由于 FastMCP 原生内置了 OpenTelemetry 埋点,只需完成 SDK 初始化即可获得遥测数据,无需在业务代码中手工打点。
初始化代码解析:otel_setup.py
遥测初始化逻辑位于 otel_setup.py,server.py 与 test_server.py 都在导入后立即调用 setup_opentelemetry(…):
def setup_opentelemetry(service_name: str) -> None:
"""Sets up OpenTelemetry to send traces to Google Cloud Observability."""
credentials, project_id = google.auth.default()
if not project_id:
raise Exception("Could not determine Google Cloud project ID.")
resource = Resource.create(
attributes={
SERVICE_NAME: service_name,
"gcp.project_id": project_id,
}
)
# Set up OTLP auth
request = google.auth.transport.requests.Request()
auth_metadata_plugin = AuthMetadataPlugin(credentials=credentials, request=request)
channel_creds = grpc.composite_channel_credentials(
grpc.ssl_channel_credentials(),
grpc.metadata_call_credentials(auth_metadata_plugin),
)
# Set up OpenTelemetry Python SDK
tracer_provider = TracerProvider(resource=resource)
tracer_provider.add_span_processor(
BatchSpanProcessor(
OTLPSpanExporter(
credentials=channel_creds,
endpoint="https://telemetry.googleapis.com:443/v1/traces",
)
)
)
trace.set_tracer_provider(tracer_provider)
logger.info("OpenTelemetry successfully initialized.")
关键实现细节(otel_setup.py):
- 通过 google.auth.default() 获取应用默认凭据与项目 ID,在 Cloud Run 环境中会自动使用服务账号身份;
- Resource.create 为遥测数据附加 service.name 与 gcp.project_id 属性,便于在 Cloud Trace 中按服务区分;
- 使用 AuthMetadataPlugin 把 Google Cloud 凭据包装为 gRPC 元数据认证插件,与 TLS 通道组合成复合通道凭据,实现 OTLP 上报的免手动令牌认证;
- OTLPSpanExporter 将 span 上报到 https://telemetry.googleapis.com:443/v1/traces(Telemetry/OTLP API 端点);
- BatchSpanProcessor 负责批量、异步地导出 span,降低对服务器性能的影响。
启用所需 API
在运行观测功能前,需要确认目标 GCP 项目已启用 Telemetry(OTLP)API、Cloud Logging API 与 Cloud Monitoring API:
gcloud services enable logging.googleapis.com monitoring.googleapis.com telemetry.googleapis.com
运行服务器并产生遥测数据
示例已经预配置好 OpenTelemetry,可直接运行:
- 本地运行:执行 uv run server.py 启动服务器,然后在另一个终端用 uv run test_server.py 调用工具产生 trace;
- Cloud Run 运行:按上文"部署"一节的命令部署即可,仓库默认的 Dockerfile 已经配置为运行插桩后的服务器(CMD ["uv", "run", "server.py"]),无需额外修改。
查看 Trace
与服务器交互产生 trace 后,即可在 Google Cloud Console 中查看。Cloud Trace 的查找与检索操作可参考 Google Cloud Trace 官方文档中"查找 traces"的指引:按服务名、项目 ID、时间范围等条件过滤,即可定位 mcp-server 每次工具调用的完整调用链与耗时。
小结
通过本示例,你可以在 python-docs-samples 仓库的 run/mcp-server 目录中完整复现"云端 MCP 服务器"的全生命周期:
这一模式将"可伸缩、集中共享、安全认证、可观测"四个能力统一到了云端,为团队协作提供了一套低成本、可复制的 MCP 服务器部署范式。
赞
【免费下载链接】python-docs-samples
Code samples used on cloud.google.com
项目地址:
https://gitcode.com/GitHub_Trending/py/python-docs-samples
点击查看 免费下载
相关推荐
Mac终极NTFS读写解决方案:Nigate工具完整使用指南
APIs-made-in-Iran人工智能API:波斯语NLP与OCR技术完全手册
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考
网硕互联帮助中心




评论前必须登录!
注册