基于FastMCP框架开发智能问答应用

示例代码

📎 附件：fastmcp-demo.zip（zip, 137KB）

原文链接：https://www.yuque.com/attachments/yuque/0/2025/zip/2424104/1756179542892-c7318315-e562-4405-b52a-362232f3397d.zip

MCP模型上下文协议实现规范

理论知识

模型上下文协议（Model Context Protocol，MCP）是由Anthropic在2024年11月推出的开放标准协议，旨在为大语言模型（LLM）提供标准化的上下文连接。MCP解决了AI应用与外部数据源和工具集成时的安全性、可扩展性和标准化问题。

MCP的核心价值：

• 标准化接口：为LLM应用提供统一的外部系统连接方式
• 安全性保障：内置安全机制，支持OAuth2.0认证和授权
• 可扩展架构：支持多种传输协议和数据格式
• 企业级应用：适合生产环境的安全和性能要求

​

最佳实践

1. 安全优先原则

• 实现robust同意和授权流程
• 提供清晰的安全影响文档
• 实施适当的访问控制和数据保护
• 遵循安全集成最佳实践

1. 协议实现规范

• 生产环境MCP服务器应支持多种传输类型
• 必须实施PKCE流程
• 实现OAuth发现端点（/.well-known/oauth-protected-resource）

示例代码

from mcp.server.fastmcp import FastMCP
from mcp.server.models import InitializationOptions
import asyncio

# 基础MCP服务器初始化
mcp = FastMCP("demo-server")

@mcp.tool()
def get_server_info() -> dict:
    """获取服务器基本信息"""
    return {
        "name": "Demo MCP Server",
        "version": "1.0.0",
        "protocol": "MCP",
        "description": "演示用MCP服务器"
    }

async def main():
    """启动MCP服务器"""
    await mcp.run()

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

MCP消息协议和传输模式原理

理论知识

MCP采用客户端-服务器架构，支持多种传输协议：

1. Stdio传输：通过标准输入输出进行通信，适合本地应用
2. SSE传输：基于服务器发送事件，适合Web应用和远程连接
3. WebSocket传输：双向实时通信，适合交互式应用

消息协议基于JSON-RPC 2.0标准，包含：

• 请求消息：客户端向服务器发送的操作请求
• 响应消息：服务器对请求的回复
• 通知消息：单向消息，不需要响应

最佳实践

1. 传输协议选择

• 本地集成优选Stdio
• Web应用使用SSE
• 实时交互选择WebSocket

1. 消息处理规范

• 实现完整的错误处理机制
• 支持消息超时和重试
• 维护连接状态管理

示例代码

from mcp.server.fastmcp import FastMCP
from mcp.types import TextContent, ImageContent
import json
import asyncio

mcp = FastMCP("message-demo")

@mcp.tool()
def process_text_message(content: str) -> dict:
    """处理文本消息"""
    return {
        "message_type": "text",
        "processed_content": content.upper(),
        "timestamp": "2024-12-01T00:00:00Z",
        "status": "processed"
    }

@mcp.resource("messages://recent")
async def get_recent_messages() -> str:
    """获取最近消息资源"""
    messages = [
        {"id": 1, "content": "Hello MCP", "type": "text"},
        {"id": 2, "content": "Protocol Demo", "type": "text"}
    ]
    return json.dumps(messages, indent=2)

# SSE传输配置示例
async def start_sse_server():
    """启动SSE传输模式服务器"""
    from mcp.server.sse import create_sse_server
    
    server = create_sse_server(mcp)
    await server.serve_forever()