getInfoTest.py•5.69 kB
import argparse
import json
import httpx
import uvicorn
from mcp.server import FastMCP, Server
from mcp.server.sse import SseServerTransport
from starlette.applications import Starlette
from starlette.routing import Route, Mount
mcp = FastMCP("sse_WeatherServer")
QWEATHER_API_KEY = "05a3e2c04b65416e912088b76a7a487e"
# 城市搜索API地址
QWEATHER_CITY_LOOKUP_URL = "https://geoapi.qweather.com/v2/city/lookup"
# 实时天气API地址
QWEATHER_BASE_URL = "https://devapi.qweather.com/v7/weather/now"
USER_AGENT = "weather-app/1.0"
async def get_location_id(city):
"""
从和风天气API获取城市的locationId
:param city: 城市名称（中文或英文）
:return: 城市的locationId；若发生错误，返回None
"""
params = {
"location": city,
"key": QWEATHER_API_KEY,
"lang": "zh"
}
headers = {"User-Agent": USER_AGENT}
async with httpx.AsyncClient() as client:
try:
response = await client.get(QWEATHER_CITY_LOOKUP_URL, params=params, headers=headers, timeout=30)
response.raise_for_status()
data = response.json()
if data.get("code") == "200" and data.get("location") and len(data["location"]) > 0:
return data["location"][0]["id"], data["location"][0]["name"]
else:
return None, None
except Exception as e:
return None, None
async def get_weather(city):
"""
从和风天气API获取天气信息
:param city: 城市名称（中文或英文）
:return: 天气数据字典；若发生错误，返回包含error信息的字典
"""
# 先获取城市的locationId
location_id, location_name = await get_location_id(city)
if not location_id:
return {"error": f"无法找到城市'{city}'的位置ID"}
params = {
"location": location_id, # 使用locationId而不是城市名
"key": QWEATHER_API_KEY,
"lang": "zh",
"unit": "m"
}
headers = {"User-Agent": USER_AGENT}
async with httpx.AsyncClient() as client:
try:
response = await client.get(QWEATHER_BASE_URL, params=params, headers=headers, timeout=30)
response.raise_for_status()
data = response.json()
# 添加城市名称到返回数据中
if "now" in data:
data["city_name"] = location_name
return data
except httpx.HTTPStatusError as e:
return {"error": f"HTTP请求错误：{e}"}
except Exception as e:
return {"error": f"发生错误：{e}"}
def format_weather_data(data):
"""
格式化天气数据
:param data: 天气数据字典
:return: 格式化后的字符串；若发生错误，返回包含error信息的字符串
"""
# 如果传入的是字符串，则先转换成字典
if isinstance(data, str):
data = json.loads(data)
if "error" in data:
return data["error"]
# 检查API返回的状态码
if data.get("code") != "200":
return f"获取天气数据失败，错误码：{data.get('code')}"
now = data["now"]
city = data.get("city_name", "未知城市")
weather = now["text"]
temperature = now["temp"]
humidity = now["humidity"]
wind_dir = now["windDir"]
wind_speed = now["windSpeed"]
return f"城市：{city}\n天气：{weather}\n温度：{temperature}°C\n湿度：{humidity}%\n风向：{wind_dir}\n风速：{wind_speed}km/h"
@mcp.tool(
name="get_weather_tool",
description="""获取指定城市的实时天气信息，包括温度、湿度、天气状况、风向和风速。
该工具使用两步查询流程：
1. 首先通过城市名称获取城市的精确位置ID
2. 然后使用位置ID查询实时天气数据
示例用法：
- 获取"北京"的天气信息
- 获取"上海"的实时天气状况
- 查询"广州"的温度和湿度
返回格式化的天气信息文本，包含城市名称、天气状况、温度、湿度、风向和风速。"""
)
async def get_weather_tool(city: str):
"""
获取城市的天气信息
:param city: 城市名称（支持中文或英文）
:return: 格式化的天气信息字符串；若发生错误，返回包含error信息的字符串
"""
weather_data = await get_weather(city)
return format_weather_data(weather_data)
def create_starlette_app(mcp_server: Server, *, debug: bool = False):
"""创建 Starlette 应用能通过sse提供mcp服务"""
sse = SseServerTransport("/messages/")
async def handle_sse(request):
async with sse.connect_sse(
request.scope,
request.receive,
request._send,
) as (read_stream,write_stream):
await mcp_server.run(
read_stream,
write_stream,
mcp_server.create_initialization_options(),
)
return Starlette(
debug=debug,
routes=[
Route("/sse", endpoint=handle_sse),
Mount("/messages/", app=sse.handle_post_message),
],
)
if __name__ == "__main__":
mcp_server = mcp._mcp_server
parser = argparse.ArgumentParser(description='Run MCP SSE-based server')
parser.add_argument("--host", default="0.0.0.0", help="MCP server host")
parser.add_argument("--port", default=18082, type=int, help="MCP server port")
args = parser.parse_args()
starlette_app = create_starlette_app(mcp_server, debug=True)
uvicorn.run(starlette_app, host=args.host, port=args.port)