技能点MCPMCP服务器【2025】🔥MCP服务器是什么？应该怎么使用？🚀本期主要讲MCP服务器
FlashFinding一、MCP服务器概念介绍
根据MCP协议定义，Server可以提供三种类型的标准能力，Resources、Tools、Prompts，每个
Server可同时提供者三种类型能力或其中一种。
Resources：资源，类似于文件数据读取，可以是文件资源或是API响应返回的内容。
Tools：工具，第三方服务、功能函数，通过此可控制LLM可调用哪些函数。
Prompts：提示词，为用户预先定义好的完成特定任务的模板。
二、MCP服务器的通讯机制
MCP定义了Client与Server进行通讯的协议与消息格式，其支持两种类型通讯机制：标准输入
输出通讯、基于SSE的HTTP通讯，分别对应着本地与远程通讯。Client与Server间使用JSON-RPC 2.0格
式进行消息传输。
本地通讯：使用了stdio传输数据，具体流程Client启动Server程序作为子进程，其消息通讯是通过
stdin/stdout进行的，消息格式为JSON-RPC 2.0。
远程通讯：Client与Server可以部署在任何地方，Client使用SSE与Server进行通讯，消息的格式为
JSON-RPC 2.0，Server定义了/see与/messages接口用于推送与接收数据。
三、创建一个天气查询服务器
3.1 OpenWeather的API key
首先需要一个OpenWeather的API key来支持天气数据的获取。
OpenWeather官网：https://openweathermap.org/
获取API key后可以先测试查询是否可行,将YOUR_API_KEY替换成自己创建的key即可。
1
2
| curl -s "https://api.openweathermap.org/data/2.5/weather?
q=Beijing&appid='YOUR_API_KEY'&units=metric&lang=zh_cn"
|
测试无误后，接下来即可进入到创建server的环节中。
3.2 服务器依赖安装
3.3 服务器代码编写
天气查询MCP执行流程：
1
2
3
4
5
6
| MCP协议
客户端（client）—发送请求→服务器（server）
客户端（client）←进行响应—服务器（server）
↓↑ (发送请求进行响应)
HTTP请求 OpenWeather API
|
weather.py(代码):
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
| import json
import httpx
from typing import Any
from mcp.server.fastmcp import FastMCP
# 初始化 MCP 服务器
mcp = FastMCP("WeatherServer")
# OpenWeather API 配置
OPENWEATHER_API_BASE = "https://api.openweathermap.org/data/2.5/weather"
API_KEY = "YOUR_API_KEY" # 请替换为你自己的 OpenWeather API Key
USER_AGENT = "weather-app/1.0"
async def fetch_weather(city: str) -> dict[str, Any] | None:
"""
从 OpenWeather API 获取天气信息。
:param city: 城市名称（需使用英文，如 Beijing）
:return: 天气数据字典；若出错返回包含 error 信息的字典
"""
params = {
"q": city,
"appid": API_KEY,
"units": "metric",
"lang": "zh_cn"
}
headers = {"User-Agent": USER_AGENT}
async with httpx.AsyncClient() as client:
try:
response = await client.get(OPENWEATHER_API_BASE, params=params,headers=headers, timeout=30.0)
response.raise_for_status()
return response.json() # 返回字典类型
except httpx.HTTPStatusError as e:
return {"error": f"HTTP 错误: {e.response.status_code}"}
except Exception as e:
return {"error": f"请求失败: {str(e)}"}
def format_weather(data: dict[str, Any] | str) -> str:
"""
将天气数据格式化为易读文本。
:param data: 天气数据（可以是字典或 JSON 字符串）
:return: 格式化后的天气信息字符串
"""
# 如果传入的是字符串，则先转换为字典
if isinstance(data, str):
try:
data = json.loads(data)
except Exception as e:
return f"无法解析天气数据: {e}"
# 如果数据中包含错误信息，直接返回错误提示
if "error" in data:
return f"⚠️ {data['error']}"
# 提取数据时做容错处理
city = data.get("name", "未知")
country = data.get("sys", {}).get("country", "未知")
temp = data.get("main", {}).get("temp", "N/A")
humidity = data.get("main", {}).get("humidity", "N/A")
wind_speed = data.get("wind", {}).get("speed", "N/A")
# weather 可能为空列表，因此用 [0] 前先提供默认字典
weather_list = data.get("weather", [{}])
description = weather_list[0].get("description", "未知")
return (
f"🌍 {city}, {country}\n"
f"🌡 温度: {temp}°C\n"
f"💧 湿度: {humidity}%\n"
f"🌬 风速: {wind_speed} m/s\n"
f"🌤 天气: {description}\n"
)
@mcp.tool()
async def query_weather(city: str) -> str:
"""
输入指定城市的英文名称，返回今日天气查询结果。
:param city: 城市名称（需使用英文）
:return: 格式化后的天气信息
"""
data = await fetch_weather(city)
return format_weather(data)
if __name__ == "__main__":
# 以标准 I/O 方式运行 MCP 服务器
mcp.run(transport='stdio')
|
