当前位置：首页 > news >正文

使用MCP Python SDK构建面向大语言模型的上下文协议服务

news 来源：原创 2025/4/25 5:36:19

使用MCP Python SDK构建面向大语言模型的上下文协议服务

引言：重新定义LLM交互方式

Model Context Protocol（MCP）作为专为大语言模型设计的标准化协议，正在改变开发者构建LLM应用的方式。通过将上下文管理与模型交互解耦，MCP使应用能够以安全、可扩展的方式向LLM暴露数据和功能。本文将以Python SDK实现为例，深入解析如何构建符合MCP标准的服务。

核心概念解析

1. Server（服务端）

作为MCP的核心枢纽，FastMCP服务器提供协议兼容性保障和消息路由功能。通过生命周期管理支持资源初始化与清理：

from mcp.server.fastmcp import FastMCP@dataclass
class AppContext:db: Database@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:db = await Database.connect()try:yield AppContext(db=db)finally:await db.disconnect()mcp = FastMCP("My App", lifespan=app_lifespan)

2. Resources（资源）

类RESTful GET端点，用于向LLM暴露静态/动态数据：

@mcp.resource("users://{user_id}/profile")
def get_user_profile(user_id: str) -> str:return f"Profile data for user {user_id}"

3. Tools（工具）

支持LLM执行带副作用的操作，如API调用或数据处理：

@mcp.tool()
async def fetch_weather(city: str) -> str:async with httpx.AsyncClient() as client:response = await client.get(f"https://api.weather.com/{city}")return response.text

4. Prompts（提示模板）

结构化交互模板提升LLM响应质量：

@mcp.prompt()
def debug_error(error: str) -> list[base.Message]:return [base.UserMessage("I'm seeing this error:"),base.UserMessage(error),base.AssistantMessage("What have you tried so far?")]

快速构建服务

开发环境配置

# 使用uv包管理器
uv add "mcp[cli]"
# 或传统pip
pip install mcp

最小示例服务

# server.py
from mcp.server.fastmcp import FastMCPmcp = FastMCP("Demo")@mcp.tool()
def add(a: int, b: int) -> int:return a + b@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:return f"Hello, {name}!"if __name__ == "__main__":mcp.run()

调试与部署

# 开发模式实时调试
mcp dev server.py --with pandas# 集成到Claude Desktop
mcp install server.py --name "Production Server" -v API_KEY=secret

进阶开发模式

生命周期管理

通过@asynccontextmanager实现资源的安全初始化和清理：

@asynccontextmanager
async def server_lifespan(server: Server) -> AsyncIterator[dict]:db = await Database.connect()try:yield {"db": db}finally:await db.disconnect()

ASGI服务器集成

与Starlette等ASGI框架无缝整合：

from starlette.applications import Starlette
from mcp.server.fastmcp import FastMCPmcp = FastMCP("Hybrid Server")
app = Starlette(routes=[Mount('/', app=mcp.sse_app())])

实战案例：SQLite浏览器

import sqlite3
from mcp.server.fastmcp import FastMCPmcp = FastMCP("SQL Explorer")@mcp.resource("schema://main")
def get_schema() -> str:conn = sqlite3.connect("data.db")return "\n".join(row[0] for row in conn.execute("SELECT sql FROM sqlite_master"))@mcp.tool()
def query_data(sql: str) -> str:try:conn = sqlite3.connect("data.db")return str(conn.execute(sql).fetchall())except Exception as e:return f"Error: {e}"

客户端开发指南

from mcp import ClientSession
from mcp.client.stdio import stdio_clientasync with stdio_client() as (read, write):async with ClientSession(read, write) as session:await session.Initialize()tools = await session.list_tools()result = await session.call_tool("query_data", {"sql": "SELECT * FROM users"})