MCP入门笔记2: MCP服务器开发流程，并以天气查询为例

文章目录

• 前言
• 三、MCP服务器Server开发流程
• • 1. MCP服务器的基本功能
  • 2. MCP服务器的通讯模式
  • • 2.1 MCP 通信模式总览
    • 2.2 HTTP 与 SSE 的比较
    • 2.3 本地（stdio）与远程（SSE）MCP 通信流程
    • • 2.3.1本地通信（stdio）
      • 2.3.2远程通信（HTTP + SSE）
    • 2.4 消息格式与接口定义
    • 3. MCP服务器SDK基本语法规则(未完成)
    • 4. 创建一个天气查询Server
    • • 4.1 准备工作
      • 4.2 天气查询Server的创建流程
      • • 4.2.1 服务器依赖安装
        • 4.2.2 服务器代码编写
• 四、MCP客户端Client接入服务器Server流程
• • 1. 天气查询客户端client创建流程
  • • 1.1client.py的代码实现
    • 1.2client.py测试运行
• 总结

前言

上篇博客中，笔者熟悉了MCP客户端开发流程，使用client.py成功调用本地ollama。本文继续记录学习相关内容，章节数承接上篇。

三、MCP服务器Server开发流程

1. MCP服务器的基本功能

根据MCP协议定义，Server可以提供三种类型的标准能力：Resources、Tools、Prompts，每个Server可同时提供三种类型能力或其中一种。

• Resources：资源，客户端可以读取的类似文件的数据（如 API 响应或文件内容）。
• Tools：工具，第三方服务、功能函数，LLM 可以调用的函数（经用户批准）。
• Prompts：提示词，帮助用户完成特定任务的预先编写的模板。

2. MCP服务器的通讯模式

2.1 MCP 通信模式总览

MCP 支持两种互补的传输方式：1. 标准输入输出（stdio） 2. 基于 HTTP 的服务器推送事件（SSE） 两种方式分别面向本地集成场景与分布式/远程集成场景，共同满足 LLM 客户端（Client）与 MCP 服务器（Server）之间的双向通信需求，并全部采用 JSON-RPC 2.0 作为消息封装格式

2.2 HTTP 与 SSE 的比较

说明： 虽然 HTTP/1.1 可复用连接，但每次请求仍属于独立交互；而 SSE 则在同一连接上多次发送事件，更适合实时性要求高的场景

2.3 本地（stdio）与远程（SSE）MCP 通信流程

2.3.1本地通信（stdio）

客户端将 MCP 服务器作为子进程启动，二者通过操作系统的 stdin/stdout 管道进行数据交换。

每条消息都使用 JSON-RPC 2.0 格式（包含 jsonrpc、method、params、id 等字段），写入子进程的 stdin；服务器处理后将响应写入 stdout。

典型场景：桌面应用（如 Claude Desktop）在本地启动 MCP 服务器，以便模型读取本地文件或执行系统命令​。

2.3.2远程通信（HTTP + SSE）

客户端通过 HTTP 向 MCP 服务器的/messages 接口发送 JSON-RPC 请求；服务器返回初始 HTTP 响应后即可马上切换到 SSE 模式。

服务器在同一 HTTP 连接上，通过 SSE 持续向客户端发送后续事件（event: message），每个事件均为 JSON-RPC 格式的消息。

客户端可在任意时刻通过新的 HTTP 请求向/messages 发送下一个调用。

典型场景：分布式部署的 LLM 应用需要从远端数据库、API 获取实时上下文更新

2.4 消息格式与接口定义

MCP 统一使用 JSON-RPC 2.0 消息格式：

{
"jsonrpc": "2.0",
"id": 1,
"method": "someMethod",
"params": { /* 方法参数 */ }
}

3. MCP服务器SDK基本语法规则(未完成)

• a. 高级API调用规则
• b. 底层API调用规则

4. 创建一个天气查询Server

通过使用OpenWeather API，创建一个能够实时查询天气的服务器(server)，并使用stdio方式进行通信。在创建之前，需要完成一个准备工作。

4.1 准备工作

第一步，从官网https://openweathermap.org/，获取API-Key。进入官网点击API，然后点击API doc；下拉可以看到sign up，最后获取API-Key。 第二步，在conda中输入代码，测试一下是否可用。

curl -s "https://api.openweathermap.org/data/2.5/weather?q=Beijing&appid='YOUR_API_KEY'&units=metric&lang=zh_cn"

图片来源：https://www.bilibili.com/video/BV1NLXCYTEbj/?p=6&share_source=copy_web&vd_source=659441755107f4432b57296f459746bb

笔者使用刚注册好的api key，遇到了错误401。查看其原因，发现这个api key必须邮箱验证之后，才能用（刚注册的时候有提示，进我垃圾邮箱了，我没注意）。。。另外需要注意的是，'YOUR_API_KEY’不要保留引号，只要数字。至于其他错误情况，我暂时没遇到过。

4.2 天气查询Server的创建流程

4.2.1 服务器依赖安装

由于我们需要使用http请求来查询天气，因此需要在当前虚拟环境中添加如下依赖：

uv add mcp httpx

4.2.2 服务器代码编写

本小节创建服务器代码，MCP基本执行流程如下： 图片来源：https://www.bilibili.com/video/BV1NLXCYTEbj/?p=7&share_source=copy_web&vd_source=659441755107f4432b57296f459746bb 参考上面的链接，server.py代码如下：

from typing import Any
import httpx
from mcp.server.fastmcp import FastMCP
import json

# Initialize FastMCP server
mcp = FastMCP("weatherServer")

# API配置
NWS_API_BASE = "https://api.openweathermap.org/data/2.5/weather"
USER_AGENT = "weather-app/1.0"
API_KEY = "YOUR_API_KEY"

async def make_request(city: str) –> dict[str, Any] | None:
"""Make a request to the API with proper error handling.
从 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(NWS_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 make_request(city)
return format_weather(data)

if __name__ == "__main__":
# Initialize and run the server
mcp.run(transport='stdio')

四、MCP客户端Client接入服务器Server流程

1. 天气查询客户端client创建流程

1.1client.py的代码实现

import asyncio
import os
import json
from openai import OpenAI
from dotenv import load_dotenv
from contextlib import AsyncExitStack

from typing import Optional
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

load_dotenv() # load environment variables from .env

class MCPClient:
def __init__(self):
# Initialize session and client objects
self.exit_stack = AsyncExitStack()
self.openai_api_key = os.getenv("OPENAI_API_KEY")
self.base_url = os.getenv("BASE_URL")
self.model = os.getenv("MODEL")
if not self.openai_api_key:
raise ValueError("openai_api_key is null, 请在.env文件中设置")

self.client = OpenAI(api_key=self.openai_api_key, base_url=self.base_url)
self.session: Optional[ClientSession] = None

async def connect_to_server(self, server_script_path: str):
"""Connect to an MCP server 连接到MCP服务器并列出可用工具
Args:
server_script_path: Path to the server script (.py or .js)
"""
is_python = server_script_path.endswith('.py')
is_js = server_script_path.endswith('.js')
if not (is_python or is_js):
raise ValueError("Server script must be a .py or .js file")

command = "python" if is_python else "node"
server_params = StdioServerParameters(
command=command,
args=[server_script_path],
env=None
)

stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
self.stdio, self.write = stdio_transport
self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))

await self.session.initialize()

# List available tools
response = await self.session.list_tools()
tools = response.tools
print("\\nConnected to server with tools:", [tool.name for tool in tools])

async def process_query(self, query: str) –> str:
"""使用LLM(如qwq:32b)处理用户query并调用可用的工具(Function Calling)"""
messages = [{"role": "user", "content": query}]

response = await self.session.list_tools()
available_tools = [{
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"input_schema": tool.inputSchema
}
} for tool in response.tools]
# print(available_tools)

# 调用 OpenAI API
response = self.client.chat.completions.create(
model = self.model,
messages = messages,
tools = available_tools
)

# 处理返回的内容
content = response.choices[0]
if content.finish_reason == "tool_calls":
# 如果需要使用工具，就解析工具
tool_call = content.message.tool_calls[0]
tool_name = tool_call.function.name
tool_args = json.loads(tool_call.function.arguments)

# 执行工具
result = await self.session.call_tool(tool_name, tool_args)
print(f"\\n\\n[Calling tool {tool_name} with args {tool_args}]\\n\\n")

# 将模型返回的调用哪个工具数据和工具执行完成后的数据都存入messages中
messages.append(content.message.model_dump())
messages.append({
"role": "tool",
"content": result.content[0].text,
"tool_call_id": tool_call.id,
})

# 将上面的结果再返回给大模型用于生产最终的结果
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
)
return response.choices[0].message.content

return content.message.content

async def chat_loop(self):
"""交互式聊天循环"""
print("\\nMCP Client Started!")
print("Type your queries or 'quit' to exit.")

while True:
try:
query = input("\\nQuery: ").strip()
if query.lower() == 'quit':
break

response = await self.process_query(query)
print("\\n" + response)

except Exception as e:
print(f"\\nError: {str(e)}")

async def cleanup(self):
"""清理资源"""
await self.exit_stack.aclose()

async def main():
if len(sys.argv) < 2:
print("Usage: python client.py <path_to_server_script>")
sys.exit(1)

client = MCPClient()
try:
await client.connect_to_server(sys.argv[1])
await client.chat_loop()
finally:
await client.cleanup()

if __name__ == "__main__":
import sys
asyncio.run(main())

1.2client.py测试运行

进入之前创建的虚拟环境中，运行代码并提问：北京今天天气如何？结果如下。这里有两个地方要注意，第一，我这里用的模型是qwq32b，deepseek-r1:70b模型不行，不支持。第二，在server4.py文件中，别忘了给api值，这次可以有引号了。

uv run client4.py server4.py

他的回答对吗？还真对！ 我百度查了一下，截图如下：

总结

本文主要记录了MCP服务器开发流程，并实现了天气查询功能。不过感觉，OpenWeather API这个天气插件很成熟了。如果想添加其他新功能，有一定挑战性。可以学习国内大厂，阿里云百炼、字节Coze等产品。