MCP Server 实现一个 天气查询

news2025/4/1 6:47:01

Step1. 环境配置

安装 uv

curl -LsSf https://astral.sh/uv/install.sh | sh

Question: 什么是 uv 呢和 conda 比有什么区别?
Answer: 一个用 Rust 编写的超快速 (100x) Python 包管理器和环境管理工具,由 Astral 开发。定位为 pip 和 venv 的替代品,专注于速度、简单性和现代 Python 工作流。

创建项目目录

uv init mcp_server_test
cd mcp_server_test

创建虚拟环境并激活

uv venv
source .venv/bin/activate

安装依赖包

uv add "mcp[cli]" httpx requests

Step2. 实现 MCP Server

创建 weather.py

import requests
from mcp.server.fastmcp import FastMCP


# 初始化 MCP 服务器
mcp = FastMCP("WeatherServer")

HEWEATHER_API_KEY = "你的 key"  # ← 填入你的和风天气Key 去https://dev.qweather.com/这里申请

def get_city_id(city_name: str) -> str:
    """根据中文城市名获取和风天气 location ID"""
    url = "https://geoapi.qweather.com/v2/city/lookup"
    params = {
        "location": city_name,
        "key": HEWEATHER_API_KEY
    }
    response = requests.get(url, params=params)
    data = response.json()
    if data.get("code") == "200" and data.get("location"):
        print(data)
        return data["location"][0]["id"]
    else:
        raise ValueError(f"找不到城市: {city_name},错误信息: {data}")


def get_weather(city_name: str) -> str:
    """根据城市中文名返回当前天气中文描述"""
    try:
        location_id = get_city_id(city_name)
        url = "https://devapi.qweather.com/v7/weather/now"
        params = {
            "location": location_id,
            "key": HEWEATHER_API_KEY
        }
        response = requests.get(url, params=params)
        data = response.json()
        if data.get("code") != "200":
            return f"天气查询失败:{data.get('code')}"
        now = data["now"]
        return (
            f"🌍 城市: {city_name}\n"
            f"🌤 天气: {now['text']}\n"
            f"🌡 温度: {now['temp']}°C\n"
            f"💧 湿度: {now['humidity']}%\n"
            f"🌬 风速: {now['windSpeed']} m/s\n"
        )
    except Exception as e:
        return f"查询出错:{str(e)}"

@mcp.tool('query_weather', '查询城市天气')
def query_weather(city: str) -> str:
    """
        输入指定城市的中文名称,返回当前天气查询结果。
        :param city: 城市名称
        :return: 格式化后的天气信息
        """
    return get_weather(city)


if __name__ == "__main__":
    # 以标准 I/O 方式运行 MCP 服务器
    mcp.run(transport='stdio')

Step3. 测试 MCP Server

运行测试

mcp dev weather.py

看到下面输出就启动成功了

mcp dev weather.py
Starting MCP inspector...
Proxy server listening on port 3000

🔍 MCP Inspector is up and running at http://localhost:5173 🚀

请添加图片描述

Step4. 修改MCP Server 做成 SSE 服务

import mcp.types as types
import requests
import uvicorn
from mcp.server.lowlevel import Server
from mcp.server.sse import SseServerTransport
from starlette.applications import Starlette
from starlette.routing import Mount, Route

# ================================
# 1) 你的和风天气API Key
# ================================
HEWEATHER_API_KEY = ""  # ← 填入你的和风天气Key(例如 "abc123xxxxxx")


# ================================
# 2) 查询天气核心逻辑
# ================================
def get_city_id(city_name: str) -> str:
    """根据中文城市名获取和风天气 location ID"""
    url = "https://geoapi.qweather.com/v2/city/lookup"
    params = {
        "location": city_name,
        "key": HEWEATHER_API_KEY
    }
    response = requests.get(url, params=params)
    data = response.json()
    if data.get("code") == "200" and data.get("location"):
        # 如果成功找到城市
        return data["location"][0]["id"]
    else:
        raise ValueError(f"找不到城市: {city_name},错误信息: {data}")


def get_weather(city_name: str) -> str:
    """根据城市中文名返回当前天气(中文描述、温度、湿度、风速)"""
    try:
        location_id = get_city_id(city_name)
        url = "https://devapi.qweather.com/v7/weather/now"
        params = {
            "location": location_id,
            "key": HEWEATHER_API_KEY
        }
        response = requests.get(url, params=params)
        data = response.json()
        if data.get("code") != "200":
            return f"天气查询失败:{data.get('code')}"

        now = data["now"]
        return (
            f"🌍 城市: {city_name}\n"
            f"🌤 天气: {now['text']}\n"
            f"🌡 温度: {now['temp']}°C\n"
            f"💧 湿度: {now['humidity']}%\n"
            f"🌬 风速: {now['windSpeed']} m/s\n"
        )
    except Exception as e:
        return f"查询出错:{str(e)}"


# ================================
# 3) MCP Server 定义
# ================================
app = Server("mcp-weather")

# (A) 工具调用处理器:根据工具名称选择执行逻辑
@app.call_tool()
async def call_tool_handler(
    name: str, arguments: dict
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
    """
    MCP 工具调用处理器
    """
    if name == "query_weather":
        if "city" not in arguments:
            raise ValueError("Missing required argument 'city'")
        # 调用上面封装好的 get_weather
        weather_info = get_weather(arguments["city"])
        return [types.TextContent(type="text", text=weather_info)]
    else:
        raise ValueError(f"Unsupported tool name: {name}")


# (B) 工具列表:告知 MCP 端都有哪些可调用的工具
@app.list_tools()
async def list_tools() -> list[types.Tool]:
    """
    定义可用的 MCP 工具列表
    """
    return [
        types.Tool(
            name="query_weather",
            description="查询指定城市天气信息(基于和风天气API)",
            inputSchema={
                "type": "object",
                "required": ["city"],
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "要查询的城市名(中文)"
                    }
                },
            },
        ),
    ]


# ================================
# 4) SSE + Starlette 路由
# ================================
sse = SseServerTransport("/messages/")

async def handle_sse(request):
    """处理 /sse 路由的 SSE 连接,并将其接入 MCP Server。"""
    async with sse.connect_sse(
        request.scope, request.receive, request._send
    ) as streams:
        # 运行 MCP 应用,处理输入输出
        await app.run(
            streams[0], streams[1], app.create_initialization_options()
        )

starlette_app = Starlette(
    debug=True,
    routes=[
        Route("/sse", endpoint=handle_sse),
        Mount("/messages/", app=sse.handle_post_message),
    ],
)


# ================================
# 5) 启动服务器
# ================================
if __name__ == "__main__":
    uvicorn.run(starlette_app, host="127.0.0.1", port=8081)

Step5. 配置 Cherry Studio

安装 Cherry Studio

https://cherry-ai.com/

配置模型 api

https://api.baystoneai.com/

请添加图片描述

这里需要选择 工具 推理 这两项

请添加图片描述

然后是在这里选择你的 MCP 服务

请添加图片描述

接下来就可以对话啦,让我们看看测试效果

请添加图片描述

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.coloradmin.cn/o/2324836.html

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!

相关文章

Headless Chrome 优化:减少内存占用与提速技巧

在当今数据驱动的时代,爬虫技术在各行各业扮演着重要角色。传统的爬虫方法往往因为界面渲染和资源消耗过高而无法满足大规模数据采集的需求。本文将深度剖析 Headless Chrome 的优化方案,重点探讨如何利用代理 IP、Cookie 和 User-Agent 设置实现内存占用…

知识就是力量——HELLO GAME WORD!

你好!游戏世界! 简介环境配置前期准备好文章介绍创建头像小功能组件安装本地中文字库HSV颜色空间音频生成空白的音频 游戏UI开发加载动画注册登录界面UI界面第一版第二版 第一个游戏(贪吃蛇)第二个游戏(俄罗斯方块&…

电脑连不上手机热点会出现的小bug

一、问题展示 注意: 不要打开 隐藏热点 否则他就会在电脑上 找不到自己的热点 二、解决办法 把隐藏热点打开即可

JAVA反序列化深入学习(八):CommonsCollections6

与CC5相似: 在 CC5 中使用了 TiedMapEntry#toString 来触发 LazyMap#get在 CC6 中是通过 TiedMapEntry#hashCode 来触发 LazyMap#get 之前看到了 hashcode 方法也会调用 getValue() 方法然后调用到其中 map 的 get 方法触发 LazyMap,那重点就在于如何在反…

鸿蒙项目源码-外卖点餐-原创!原创!原创!

鸿蒙外卖点餐外卖平台项目源码含文档包运行成功ArkTS语言。 我半个月写的原创作品,请尊重原创。 原创作品,盗版必究!!! 原创作品,盗版必究!!! 原创作品,盗版…

React程序打包与部署

===================== 推荐超级课程: 本地离线DeepSeek AI方案部署实战教程【完全版】Docker快速入门到精通Kubernetes入门到大师通关课AWS云服务快速入门实战目录 为生产环境准备React应用最小化和打包环境变量错误处理部署到托管服务部署到Netlify探索高级主题:Hooks、Su…

Leetcode算法方法总结

1. 双指针法解决链表/数组题目 只要数组有序,就要想到双指针做法。还有二分法 回文串一般也会用到双指针,回文串的长度由于可能是奇数也可能是偶数,所以在寻找时,既需要寻找奇数长度的回文串,也需要寻找偶数长度的回文…

全包圆玛奇朵样板间亮相,极简咖啡风引领家装新潮流

在追求品质生活的当下,家居装修风格的选择成为了许多消费者关注的焦点。近日,全包圆家居装饰有限公司精心打造的玛奇朵样板间正式对外开放,以其独特的咖啡色系极简风格,为家装市场带来了一股清新的潮流。玛奇朵样板间不仅展示了全…

大数据学习(92)-spark详解

🍋🍋大数据学习🍋🍋 🔥系列专栏: 👑哲学语录: 用力所能及,改变世界。 💖如果觉得博主的文章还不错的话,请点赞👍收藏⭐️留言📝支持一…

免费下载 | 2025年网络安全报告

报告总结了2024年的网络安全态势,并对2025年的安全趋势进行了预测和分析。报告涵盖了勒索软件、信息窃取软件、云安全、物联网设备安全等多个领域的安全事件和趋势,并提供了安全建议和最佳实践。 一、报告背景与目的 主题:2024企业信息安全峰…

RCE--解法

目录 一、利用php伪协议 1.代码分析 2.过程 3.结果 ​编辑 4.防御手段 二、RCE(php中点的构造&#xff09; 1.代码分析 2.过程 一、利用php伪协议 <?php error_reporting(0); if(isset($_GET[c])){$c $_GET[c];if(!preg_match("/flag|system|php|cat|sort…

JAVA反序列化深入学习(九):CommonsCollections7与CC链总结

CC7 依旧是寻找 LazyMap 的触发点 CC6使用了 HashSet而CC6使用了 Hashtable JAVA环境 java version "1.8.0_74" Java(TM) SE Runtime Environment (build 1.8.0_74-b02) Java HotSpot(TM) 64-Bit Server VM (build 25.74-b02, mixed mode) 依赖版本 Apache Commons …

HTML元素小卖部:表单元素 vs 表格元素选购指南

刚学HTML的同学经常把表单和表格搞混&#xff0c;其实它们就像超市里的食品区和日用品区——虽然都在同一个超市&#xff0c;但用途完全不同。今天带你3分钟分清这两大元素家族&#xff01; 一、表单元素家族&#xff08;食品区&#xff1a;收集用户输入&#xff09; 1. <i…

群体智能优化算法-算术优化算法(Arithmetic Optimization Algorithm, AOA,含Matlab源代码)

摘要 算术优化算法&#xff08;Arithmetic Optimization Algorithm, AOA&#xff09;是一种新颖的群体智能优化算法&#xff0c;灵感来源于加、减、乘、除四种基本算术运算。在优化过程中&#xff0c;AOA 通过乘除操作实现全局探索&#xff0c;通过加减操作强化局部开发&#…

Linux之数据链路层

Linux之数据链路层 一.以太网1.1以太网帧格式1.2MAC地址1.3MTU 二.ARP协议2.1ARP协议工作流程2.2ARP协议格式 三.NAT技术四.代理服务4.1正向代理4.2反向代理 五.四大层的学习总结 一.以太网 在我们学习完了网络层后我们接下来就要进入数据链路层的学习了&#xff0c;在学习完网…

如何在 vue 渲染百万行数据,vxe-table 渲染百万行数据性能对比,超大量百万级表格渲染

vxe-table 渲染百万行数据性能对比&#xff0c;超大量百万级表格渲染&#xff1b;如何在 vue 渲染百万行数据&#xff1b;当在开发项目时&#xff0c;遇到需要流畅支持百万级数据的表格时&#xff0c; vxe-table 就可以非常合适了&#xff0c;不仅支持强大的功能&#xff0c;虚…

MySQL-5.7.37安装配置(Windows)

1.下载MySQL-5.7.37软件包并解压 2.配置本地环境变量 打开任务栏 搜索高级系统设置 新建MySQL的环境变量 然后在path中添加%MYSQL_HOME%\bin 3.在MySQL-5.7.37解压的文件夹下新建my.ini文件并输入以下内容 [mysqld]#端口号port 3306#mysql-5.7.27-winx64的路径basedirC:\mysq…

鸿蒙北向应用开发:deveco 5.0 kit化文件相关2

鸿蒙北向应用开发:deveco 5.0 kit化文件相关 在kit化时,有时候会出现这样一种场景即你想把已有的d.ts导出换个名字,这样从名字上更贴合你的kit聚合 什么意思呢?比如现在有 ohos.hilog.d.ts 导出了hilog,现在你想kit化hilog,使得hilog导出名字为usrhilog,这样用户在使用你的k…

《HelloGitHub》第 108 期

兴趣是最好的老师&#xff0c;HelloGitHub 让你对开源感兴趣&#xff01; 简介 HelloGitHub 分享 GitHub 上有趣、入门级的开源项目。 github.com/521xueweihan/HelloGitHub 这里有实战项目、入门教程、黑科技、开源书籍、大厂开源项目等&#xff0c;涵盖多种编程语言 Python、…