用 MCP 协议统一管理 Google Cloud深入解析 Klavis 仓库中的 Google Cloud MCP Server【免费下载链接】klavisKlavis AI: MCP integration platforms that let AI agents use tools reliably at any scale项目地址: https://gitcode.com/GitHub_Trending/kl/klavis本篇技术指南以 Klavis 仓库GitHub_Trending/kl/klavis内收录的mcp_servers/google_cloud_toolathlon项目为主体系统讲解如何基于 Model Context ProtocolMCP构建一个覆盖 BigQuery、Cloud Logging、Cloud Storage、Compute Engine 四大 GCP 服务的统一 AI 工具服务器。读完本文你将掌握该 MCP Server 的安装启动、两种认证方式、四大服务共数十个工具的参数细节、Claude Desktop 接入配置以及从源码层面理解其模块化设计、访问控制Allowlist与多传输协议SSE / StreamableHTTP的实现原理。项目概览一个面向 AI Agent 的 GCP 操作网关google_cloud_toolathlon是一个功能完整的 MCP Server为 AI 模型与 Google Cloud Platform 之间提供结构化交互接口。其核心思路是把 GCP 各服务的官方 Python SDK 封装成统一的 Manager 类再通过FastMCP注册为一个个可被 AI 调用、返回 JSON 友好字典的 MCP 工具。项目主要能力如下BigQuery数据仓库操作包括 SQL 查询执行含 dry-run 成本估算、数据加载/导出、作业Job管理与取消Cloud Logging日志读写、日志桶Log Bucket管理、Sink 导出、Exclusion 过滤规则Cloud Storage桶与对象的完整 CRUD、生命周期策略、批量操作、签名 URLSigned URL生成Compute Engine虚拟机生命周期管理创建/删除/启动/停止/重启与区域Zone查询MCP 集成完整实现 MCP 协议支持 SSE 与 StreamableHTTP 双传输灵活认证同时支持服务账号 JSON 文件与默认凭据ADC并支持基于 OAuth Token 的按请求认证。安装与启动从 uv 工具安装项目在 PyPI 上以google-cloud-mcp发布可通过 uv 直接安装uv tool install google-cloud-mcp从源码运行在当前仓库中项目实际代码位于mcp_servers/google_cloud_toolathlon/入口为src/server.py。依赖清单见 requirements.txt其中关键依赖包括fastmcp2.10.6与mcp1.12.2MCP 协议与 FastMCP 框架google-cloud-bigquery3.35.1、google-cloud-storage3.2.0、google-cloud-logging3.12.1、google-cloud-compute1.33.0uvicorn0.35.0、starlette0.47.2HTTP 服务承载安装依赖后直接运行pip install -r mcp_servers/google_cloud_toolathlon/requirements.txt python -m mcp_servers.google_cloud_toolathlon.src.server \ --project-id your-project-id \ --service-account-path /path/to/service-account-key.json服务默认监听0.0.0.0:5000同时暴露两个端点SSE 端点http://localhost:5000/sseStreamableHTTP 端点http://localhost:5000/mcp认证方式项目在server.py与各 Manager 的构造函数中实现了三种凭据来源从源码看其优先级逻辑清晰按请求 OAuth 凭据最高优先级若通过请求头x-auth-data或环境变量AUTH_DATA携带 OAuth 信息access_token、refresh_token、token_uri、client_id、client_secret 等_get_oauth_credentials()会构造OAuthCredentials为每个请求动态创建 Manager见 server.py。服务账号 JSON 文件--service-account-path参数或项目根目录的service-account-key.json。各 Manager 通过service_account.Credentials.from_service_account_file()加载scope 为https://www.googleapis.com/auth/cloud-platform。默认凭据ADC未提供任何参数时回落到 Google Cloud SDK 的Application Default Credentialsgcloud auth application-default loginProject ID 的解析顺序在 server.py 的_get_project_id()中项目 ID 按如下顺序解析每次请求的 auth data 中携带的project_id全局配置PROJECT_ID来自--project-id参数或PROJECT_ID/GOOGLE_CLOUD_PROJECT环境变量通过 Google Cloud Resource Manager API 自动解析——当账号仅有一个项目时直接返回有多个项目时优先匹配账号邮箱前缀对应的项目否则回落到第一个项目_resolve_project_id_from_api。可用工具详解以下所有工具均通过mcp.tool()注册于 server.py名称带有google-cloud-前缀。BigQuery 工具BigQuery 操作由BigQueryManagerbig_query.py支撑。查询执行的返回结构统一为包含results、total_rows、bytes_processed、bytes_billed、estimated_cost_usd、execution_time_ms、job_id的字典。执行 SQL 查询支持 dry-run 成本估算{ query: SELECT * FROM dataset.table LIMIT 10, dry_run: false, max_results: 1000 }在源码中dry_run 模式通过QueryJobConfig(dry_runTrue)实现成本估算公式为(total_bytes_billed / (1024**4)) * 5.0即按 $5/TB 计价见 big_query.py 第 87-97 行。非 dry-run 时工具响应会包含总行数、处理字节数、执行耗时并默认只展示前 5 行样本数据。创建数据集{ dataset_id: my_dataset, description: My dataset description, location: US }创建表带 Schema{ dataset_id: my_dataset, table_id: my_table, schema: [ {name: id, type: INTEGER, mode: REQUIRED}, {name: name, type: STRING, mode: NULLABLE} ] }从 CSV 加载数据{ dataset_id: my_dataset, table_id: my_table, csv_file_path: /path/to/data.csv, write_disposition: WRITE_APPEND }load_data_from_csv()底层使用bigquery.LoadJobConfig(source_formatCSV, autodetectTrue)write_disposition支持WRITE_TRUNCATE、WRITE_APPEND、WRITE_EMPTY三种取值若关闭 autodetect也可显式传入 schema 定义。导出表到 Cloud Storage{ dataset_id: my_dataset, table_id: my_table, bucket_name: my-bucket, file_path: exports/data.csv }当前导出仅支持 CSV 格式ExtractJobConfig(destination_formatCSV)且同时校验目标数据集与目标桶的访问权限。其他工具list_datasets列出数据集、list_tables列出数据集下的表、get_job_status查询作业状态、list_jobs按RUNNING/DONE/PENDING过滤、cancel_job取消作业。Cloud Logging 工具Cloud Logging 由CloudLoggingManagercloud_logging.py支撑内部同时持有logging_v2.Client与ConfigServiceV2Client前者负责读写日志后者负责桶、Sink、Exclusion 的配置管理。写入日志文本或结构化{ log_name: my-application, message: Application started successfully, severity: INFO, labels: {component: backend} }源码中字符串消息走log_text()字典消息走log_struct()默认 resource 类型为global。读取与过滤日志{ filter_string: severityERROR, max_results: 100, time_range_hours: 24 }read_logs()会自动拼接timestamp ...时间过滤与用户 filter按timestamp desc排序每条日志会解析出 severity兼容字符串与枚举、text/json/proto payload、labels、trace、span_id 与 resource 信息。搜索日志简化参数{ search_query: error occurred, time_range_hours: 12, severity_levels: [ERROR, CRITICAL] }search_logs()会把查询词包装为(textPayload:... OR jsonPayload:...)并支持 resource.type 与 severity 的 OR 组合过滤。日志桶管理create_log_bucketretention_days控制保留天数、locked防止误删、update_log_bucket、delete_log_bucket、clear_log_bucket保留桶但清空其中日志、list_log_buckets。删除或清空已锁定locked的桶会直接抛出异常。Sink 导出create_log_sink目标格式如storage.googleapis.com/bucket或bigquery.googleapis.com/projects/PROJECT/datasets/DATASET、list_log_sinks、delete_log_sink。此外还有两个便捷封装export_logs_to_storage创建指向 Cloud Storage 桶的 Sinkexport_logs_to_bigquery创建指向 BigQuery 数据集的 Sink。两者都会在返回结果中附带writer_identity需要为它授予目标资源的写权限。Exclusion 过滤create_exclusion如severityINFO过滤掉低级别日志、list_exclusions、delete_exclusion。Cloud Storage 工具Cloud Storage 由CloudStorageManagercloud_storage.py支撑。所有返回对象都经过_bucket_to_dict()/_blob_to_dict()标准化为字典便于 JSON 序列化。创建桶{ bucket_name: my-new-bucket, location: US, storage_class: STANDARD }storage_class支持STANDARD、NEARLINE、COLDLINE、ARCHIVE。上传/下载/删除/列出对象{ bucket_name: my-bucket, local_file_path: /path/to/file.txt, blob_name: uploads/file.txt }{ bucket_name: my-bucket, prefix: uploads/, max_results: 100 }桶内/跨桶复制与移动{ source_bucket: source-bucket, source_blob: file.txt, destination_bucket: dest-bucket, destination_blob: backup/file.txt }move_object在源码中是先复制再删除源对象的组合操作。生成签名 URL临时访问{ bucket_name: my-bucket, blob_name: private/file.txt, expiration_hours: 24, method: GET }底层使用 v4 签名expiration以分钟为单位计算method支持 GET/PUT/POST/DELETE。生命周期策略{ bucket_name: my-bucket, rules: [ { action: Delete, conditions: {age: 365} } ] }对应工具set_bucket_lifecycle的参数为age_days与actionDelete或SetStorageClass源码会读取现有规则列表并追加新规则后bucket.patch()。其他工具batch_upload/batch_download批量文件映射、search_files基于 fnmatch 通配符与大小范围搜索、enable_versioning、get_bucket_size统计对象数与按存储类拆分的体积、get_bucket_info。Compute Engine 工具Compute Engine 由ComputeEngineManagercompute_engine.py支撑底层使用compute_v1.InstancesClient、ZonesClient、ZoneOperationsClient。创建虚拟机实例{ instance_name: my-vm, zone: us-central1-a, machine_type: e2-medium, image_family: ubuntu-2004-lts, image_project: ubuntu-os-cloud }源码中create_instance()的默认值为machine_typee2-micro、镜像debian-11、磁盘 10GB、网络default、默认分配外网 IPONE_TO_ONE_NAT。创建、启动、停止、删除均为异步操作返回operation.name可用wait_for_operation等待完成。实例生命周期操作delete_instance、start_instance、stop_instance、restart_instance重启前会检查实例必须处于RUNNING或STOPPING状态。查询操作{ instance_name: my-vm, zone: us-central1-a }get_instance返回状态、区域、机器类型、内外网 IP、启动盘大小、网络标签与 Labelslist_instances支持按 zone 过滤或跨全区域聚合查询list_zones只返回状态为UP的区域get_operation_status/wait_for_operation用于轮询异步操作结果。访问控制Allowlist机制这是该 MCP Server 在安全设计上的关键特性。启动时可通过四个参数限定 AI 可操作的资源范围server.py 第 271-311 行--allowed-buckets允许的 Cloud Storage 桶列表--allowed-datasets允许的 BigQuery 数据集列表--allowed-log-buckets允许的日志桶列表--allowed-instances允许的 Compute Engine 实例列表每个列表均为逗号分隔且支持通配符前缀匹配_matches_allowed_pattern()规定以*结尾的模式如prefix*会匹配所有以该前缀开头的资源名否则做精确匹配server.py 第 313-338 行。未配置 Allowlist 时默认放行配置后工具会在执行前调用validate_bucket_access()/validate_dataset_access()等函数拦截越权操作例如创建不在允许列表中的数据集会直接返回Access denied。接入 Claude Desktop将以下配置加入 Claude Desktop 的 MCP settings 文件即可{ mcpServers: { google-cloud-mcp: { command: uvx, args: [ google-cloud-mcp, --project-id, your-project-id, --service-account-path, /path/to/your/service-account-key.json, --allowed-buckets, bucket1,bucket2,bucket3, --allowed-datasets, dataset1,dataset2, --allowed-log-buckets, log-bucket1, --allowed-instances, vm1,vm2 ] } } }配置要点Project ID替换your-project-id为真实 GCP 项目 IDService Account--service-account-path指向服务账号 JSON 文件访问控制用逗号分隔的 Allowlist 收窄 AI 可触达的资源范围建议始终配置。架构与源码设计模式模块化结构mcp_servers/google_cloud_toolathlon/ ├── src/ │ ├── __init__.py │ ├── server.py # MCP 服务器实现工具注册、传输、认证 │ ├── big_query.py # BigQuery Manager │ ├── cloud_logging.py # Cloud Logging Manager │ ├── cloud_storage.py # Cloud Storage Manager │ └── compute_engine.py # Compute Engine Manager ├── Dockerfile # 容器化构建python:3.12-slim ├── README.Docker.md # Docker 部署指南 ├── README.md # 项目说明 ├── CLAUDE.md # 开发指南 └── requirements.txt # 固定版本依赖Manager 类的统一模式四个 Manager 遵循一致的约定这也是理解整个代码库的钥匙构造函数接收project_id和可选的service_account_path或直接注入 credentials 对象方法统一返回字典便于 JSON 序列化与 MCP 工具包装如_bucket_to_dict()、_instance_to_dict()等私有转换方法负责标准化输出完备的错误处理针对NotFound、Conflict、AlreadyExists等 GCP 异常做分类捕获与日志记录认证链server.py通过get_*_manager()工厂函数优先返回携带每请求 OAuth 凭据的 Manager否则回落至全局初始化好的 Manager。双传输协议实现main()中默认使用 StreamableHTTP--transport streamable-http同时保留 SSE 兼容--transport stdio时走标准输入输出传输便于本地进程接入。HTTP 模式下注册了三条路由server.py 第 1542-1552 行GET /sseSSE 事件流端点POST /messages/SSE 消息回传端点Mount /mcpStreamableHTTP 端点。两条 HTTP 传输路径都会在请求处理前后通过ContextVar注入并清理按请求的 OAuth 上下文access_token、project_id、refresh_token 等实现多租户下的请求级认证隔离。容器化部署项目提供了现成的 Dockerfile基于python:3.12-slim暴露 5000 端口从仓库根目录构建docker build -f mcp_servers/google_cloud_toolathlon/Dockerfile -t google-cloud-mcp . docker run -d --name google-cloud-mcp -p 5000:5000 \ -e GOOGLE_CLOUD_PROJECTyour-project-id \ -e GOOGLE_APPLICATION_CREDENTIALS/app/service-account-key.json \ -v $(pwd)/service-account-key.json:/app/service-account-key.json:ro \ google-cloud-mcp容器化部署支持的环境变量汇总详见 README.Docker.md变量必填说明GOOGLE_CLOUD_PROJECT是GCP 项目 IDGOOGLE_APPLICATION_CREDENTIALS否服务账号密钥文件路径HOST/PORT否监听地址默认0.0.0.0:5000ALLOWED_BUCKETS否允许的 GCS 桶逗号分隔ALLOWED_DATASETS否允许的 BigQuery 数据集ALLOWED_LOG_BUCKETS否允许的日志桶ALLOWED_INSTANCES否允许的实例AUTH_DATA否Base64 编码的 OAuth 认证数据测试与开发建议项目通过uv run test_server.py遍历全部 MCP 工具以验证功能需真实 GCP 凭据与相应 API 权限服务账号建议采用最小权限原则按需授予 BigQuery Data Editor、Storage Admin、Logging Admin、Compute Admin 等角色生产环境请配合密钥管理Docker Secrets / Kubernetes Secrets、只读挂载凭据、定期轮换服务账号密钥并始终使用ALLOWED_*访问控制收窄 AI 的操作面。适用前提与限制运行环境要求 Python 3.8容器镜像使用 Python 3.12需安装requirements.txt中列出的 Google Cloud 官方 SDK使用默认凭据需要本地安装 Google Cloud SDK本文所述工具参数、默认值如 BigQuery 成本按 $5/TB 估算、Compute 默认e2-micro等均以当前仓库源码为准实际部署时请以 server.py 与各 Manager 模块的 docstring 为最终依据。【免费下载链接】klavisKlavis AI: MCP integration platforms that let AI agents use tools reliably at any scale项目地址: https://gitcode.com/GitHub_Trending/kl/klavis创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考