ARTICLE DETAIL

资讯详情

深耕编程入门与网站建设的一线实战洞察。

【案例共创】在华为开发者空间用 FastAPI 搭建 MCP 天气查询服务:TaoToken 统一 Key 配置与联调验证

【案例共创】在华为开发者空间用 FastAPI 搭建 MCP 天气查询服务:TaoToken 统一 Key 配置与联调验证 1. 从一次“天气查不到”说起为什么要在华为开发者空间搭 MCP 天气服务你在华为开发者空间里跑起了一个 Cherry Studio接上了 DeepSeek问它“北京今天天气怎么样”它一本正经地回你“我无法获取实时天气数据”。这不是模型不行而是它缺一个能查实时数据的工具。MCPModel Context Protocol就是干这个的把外部 API 包装成模型能调用的标准工具让对话从“背课文”变成“查现场”。这篇要做的是在华为开发者空间的云主机上用 FastAPI 写一个天气查询服务再通过 fastapi-mcp 把它暴露成 MCP 工具最后在 Cherry Studio 里完成一次真实的“北京天气”调用。整条链路里有个容易被忽略但很烦的点你手头可能同时有 Cherry Studio、Cline、Claude Code 好几个工具每个都要填一遍 API Key、改一遍地址改到后面自己都记不清哪个 Key 对应哪个服务。所以我会把 TaoToken 的统一 Key 通道一起接进来用一份配置管住多工具接入省掉反复复制粘贴的麻烦。适合谁看在华为开发者空间有云主机、想跑通 MCP 工具调用的开发者手头 AI 工具多、被 Key 管理搞烦的人以及想拿天气查询当模板之后换成自己业务 API 的人。下面从环境准备开始一步步给可复制的配置和命令。2. 前置准备华为开发者空间、和风天气与 TaoToken 统一 Key2.1 华为开发者空间云主机登录华为开发者空间工作台进入云主机桌面。规格用默认的鲲鹏通用计算增强型 kc2.xlarge.24vCPU/8G/Ubuntu就够这个案例本身不烧算力主要跑 FastAPI 和几个 Python 包。打开终端先确认 Python 版本python3 --version pip3 --versionUbuntu 镜像一般自带 Python 3.10 以上够用。如果 pip 版本太老先升级一下python3 -m pip install --upgrade pip -i https://repo.huaweicloud.com/repository/pypi/simple2.2 和风天气的 JWT 凭据天气数据源用和风天气。注册登录后进开发服务控制台创建项目拿到 PROJECT_ID再创建 JWT 凭据。JWT 凭据需要你先在本地生成一对 Ed25519 公私钥把公钥整段上传包括 BEGIN/END 那两行保存后拿到 KEY_ID。私钥留在云主机上给代码用。生成密钥对的命令openssl genpkey -algorithm ED25519 -out ed25519-private.pem openssl pkey -pubout -in ed25519-private.pem ed25519-public.pem然后在控制台设置页找到你的 API Host形如xxxxxx.re.qweatherapi.com。到这里PROJECT_ID、KEY_ID、私钥、API Host 四样东西齐了后面代码里要替换。2.3 TaoToken 统一 Key 通道多工具接入的痛点在于Cherry Studio 要填一套Cline 要填一套Claude Code 又要填一套每套都得记地址和 Key。TaoToken 提供的是 OpenAI 兼容的统一 API 通道你只需要在它那边生成一个 Key然后各个工具都指向同一个地址配置里只维护一份。先去控制台创建 API Keyhttps://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmcp_weather_console创建完把 Key 复制出来只显示一次。API 基础地址是https://taotoken.net/api注意这个地址不带任何查询参数直接作为 base_url 用。模型名按你实际要用的填比如gpt-4o-mini、claude-3-5-sonnet这类具体以模型列表页为准。想先验证 Key 能不能通可以用模型对话页发一条消息试试https://taotoken.net/model-chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmcp_weather_chat如果你后面要长期跑编码类 Agent可以看下 Coding Plan额度模型和按量计费不太一样https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmcp_weather_plan3. 可复制配置FastAPI fastapi-mcp 服务端骨架3.1 项目初始化与依赖安装建目录、装依赖一条命令搞定mkdir -p ~/qweather-mcp cd ~/qweather-mcp pip3 install fastapi uvicorn fastapi-mcp pyjwt cryptography requests \ -i https://repo.huaweicloud.com/repository/pypi/simple这里几个包的分工fastapi 提供 Web 框架uvicorn 是 ASGI 服务器fastapi-mcp 负责把 FastAPI 路由自动转成 MCP 工具pyjwt 和 cryptography 用来签 JWTrequests 调天气接口。3.2 服务端核心代码新建main.py把下面这份骨架贴进去。重点看三处配置区替换成你自己的值、JWT 生成逻辑、以及最后 MCP 挂载那几行。from fastapi import FastAPI, HTTPException, Depends, Query, status from fastapi.security import HTTPBearer from pydantic import BaseModel, Field from typing import List, Optional, Dict, Any import jwt import time import requests from datetime import datetime import base64 from cryptography.hazmat.primitives import serialization from cryptography.hazmat.backends import default_backend from fastapi_mcp import FastApiMCP, AuthConfig app FastAPI(title天气查询API) # 配置区替换成你自己的值 PRIVATE_KEY YOUR_PRIVATE_KEY_BASE64 PROJECT_ID YOUR_PROJECT_ID KEY_ID YOUR_KEY_ID WEATHER_API_HOST your_api_host.re.qweatherapi.com # 加载 Ed25519 私钥 try: _der_key_bytes base64.b64decode(PRIVATE_KEY) PRIVATE_KEY serialization.load_der_private_key( _der_key_bytes, passwordNone, backenddefault_backend() ) except Exception as e: print(f私钥加载失败: {e}) PRIVATE_KEY None if PRIVATE_KEY is None: raise RuntimeError(EdDSA 私钥加载失败应用无法启动) token_cache {token: None, expires_at: 0} token_auth_scheme HTTPBearer() EXPECTED_BEARER_TOKEN huaweicloud async def verify_bearer_token(token_payloadDepends(token_auth_scheme)): if token_payload.credentials ! EXPECTED_BEARER_TOKEN: raise HTTPException( status_codestatus.HTTP_401_UNAUTHORIZED, detail无效的认证凭据, headers{WWW-Authenticate: Bearer}, ) return token_payload.credentials def generate_jwt(expiry_seconds: int 900): current_time int(time.time()) payload { iat: current_time - 30, exp: current_time expiry_seconds, sub: PROJECT_ID, } headers {kid: KEY_ID} encoded_jwt jwt.encode(payload, PRIVATE_KEY, algorithmEdDSA, headersheaders) token_cache[token] encoded_jwt token_cache[expires_at] current_time expiry_seconds - 60 return encoded_jwt def get_valid_token(): current_time int(time.time()) if token_cache[token] and token_cache[expires_at] current_time: return token_cache[token] return generate_jwt() def fetch_weather_api(endpoint: str, params: Dict[str, Any]): token get_valid_token() headers { Authorization: fBearer {token}, Accept-Encoding: gzip, } url fhttps://{WEATHER_API_HOST}/{endpoint} response requests.get(url, headersheaders, paramsparams) if response.status_code in [401, 403]: headers[Authorization] fBearer {generate_jwt()} response requests.get(url, headersheaders, paramsparams) if response.status_code ! 200: raise HTTPException( status_coderesponse.status_code, detailf天气API请求失败: HTTP {response.status_code}, ) return response.json() class WeatherQuery(BaseModel): city: str Field(..., description城市名如北京) app.post(/weather/by-city, operation_idget_weather_by_city, tags[天气查询]) async def get_weather_by_city(query: WeatherQuery): location_data fetch_weather_api(geo/v2/city/lookup, {location: query.city}) if location_data.get(code) ! 200 or not location_data.get(location): raise HTTPException(status_code404, detailf找不到城市: {query.city}) location_id location_data[location][0][id] return fetch_weather_api(v7/weather/now, {location: location_id}) # MCP 挂载 mcp FastApiMCP( app, nameMy Weather MCP, description天气查询API, include_operations[get_weather_by_city], auth_configAuthConfig(dependencies[Depends(verify_bearer_token)]), ) mcp.mount() if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8009)几个关键点解释一下。include_operations只把get_weather_by_city这一个接口暴露成 MCP 工具其他路由不对外避免工具列表太乱。AuthConfig里挂了verify_bearer_token意味着 MCP 客户端连接时必须带Authorization: Bearer huaweicloud这个头否则连不上。JWT 缓存提前 60 秒过期防止边界时间戳导致 401。3.3 TaoToken 统一 Key 的 config.toml 与 settings.json如果你用 Claude Code 或类似支持配置文件的方式接入可以准备一份config.toml把 TaoToken 的地址和 Key 写进去多个项目共用# ~/.config/taotoken/config.toml [provider] name taotoken base_url https://taotoken.net/api api_key sk-你的TaoTokenKey model gpt-4o-mini [mcp.weather] name 和风天气MCP type sse url http://localhost:8009/mcp headers { Authorization Bearer huaweicloud }Cherry Studio 这类图形工具走的是settings.json风格等价配置长这样{ providers: [ { name: taotoken, type: openai, base_url: https://taotoken.net/api, api_key: sk-你的TaoTokenKey, models: [gpt-4o-mini] } ], mcpServers: { qweather: { type: sse, url: http://localhost:8009/mcp, headers: { Authorization: Bearer huaweicloud } } } }这样一份配置里模型通道和 MCP 工具通道都齐了。换工具时只改外层Key 和地址不用动。想管理或轮换 Key去 API Keys 页面https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmcp_weather_keys4. 启动与验证一次真实的北京天气查询4.1 启动服务在云主机终端里cd ~/qweather-mcp python3 main.py看到Uvicorn running on http://0.0.0.0:8009就说明起来了。先别急着接 MCP用 curl 直接打一下 REST 接口确认天气数据能拿到curl -X POST http://localhost:8009/weather/by-city \ -H Content-Type: application/json \ -d {city: 北京}正常返回里会有code: 200、now.temp、now.text这些字段。如果这一步就报 401说明 JWT 签名或 KEY_ID 有问题报 404 说明城市名没匹配上换个“北京市”试试。4.2 验证 MCP 端点MCP 走的是 SSE直接 curl 会挂住用带超时的请求探一下握手curl -N -H Authorization: Bearer huaweicloud \ --max-time 5 http://localhost:8009/mcp能看到event: endpoint之类的 SSE 事件流就说明 MCP 服务端在正常工作。如果返回 401检查请求头里的 Bearer 值是不是huaweicloud和代码里EXPECTED_BEARER_TOKEN保持一致。4.3 在 Cherry Studio 里完成工具调用打开 Cherry Studio设置里添加 MCP 服务器类型选 SSEURL 填http://localhost:8009/mcp请求头加Authorization: Bearer huaweicloud。保存后工具列表里应该出现get_weather_by_city。新建话题模型选你通过 TaoToken 接入的那个先不启用 MCP问“北京天气怎么样”模型会说无法获取实时数据。然后启用和风天气 MCP再问一次这次它会调用get_weather_by_city返回北京当前的温度、天气现象、湿度等。看到工具调用卡片展开、参数是{city: 北京}就说明整条链路打通了。5. 本篇常见错排查私钥加载失败启动直接 RuntimeError。最常见的原因是 PRIVATE_KEY 没有做 base64 解码或者复制时漏了字符。代码里是先base64.b64decode再load_der_private_key所以你填进去的应该是 base64 字符串不是 PEM 文件原文。如果你手上是 PEM 格式用base64 -w0 ed25519-private.pem转一下再填。MCP 客户端连不上报 401 或直接断开。检查三处请求头里的 Bearer 值、代码里的EXPECTED_BEARER_TOKEN、以及 URL 是不是/mcp结尾。SSE 类型对 URL 路径敏感写成/mcp/有时也会出问题。天气接口返回 401 但 JWT 看着没过期。和风天气的 JWT 对iat和exp的时间窗口比较敏感云主机如果时区或时间有偏差会直接拒。代码里iat提前了 30 秒就是防这个。如果还不行date看一下云主机时间必要时同步一下。城市查不到返回 404。geo/v2/city/lookup对中文城市名匹配有时要带“市”字或者用拼音。先单独调 lookup 接口确认能返回 location 数组再走完整流程。Cherry Studio 里工具列表为空。确认服务端日志有没有收到 SSE 连接请求。如果服务端没日志多半是 URL 或请求头不对如果有连接但没工具检查include_operations里的 operation_id 是否和路由装饰器里写的一致。6. 把天气换成你自己的 API这套骨架的价值不在天气本身而在于它示范了“REST API → MCP 工具”的转换路径。你只要把fetch_weather_api换成自己的业务接口把get_weather_by_city换成你的 operation_idMCP 挂载那几行不用动工具就能被模型调用。多工具接入的 Key 管理用 TaoToken 统一通道之后config.toml 和 settings.json 里只维护一份 base_url 和 Key换模型、加工具都只改局部。接入文档在这里里面有各语言 SDK 的调用示例https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmcp_weather_doc如果你用的是 Claude Code 这类编码 AgentAnthropic 兼容通道的配置方式单独有一页说明https://taotoken.net/claude-code-anthropic?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmcp_weather_claude最后留一个实操建议把main.py里的配置项抽到.env文件用python-dotenv读进来这样私钥和 Key 不会跟着代码进 Git。云主机上跑长期服务的话用nohup python3 main.py 或者写个 systemd unit别让终端一关服务就断。
返回列表