🚀 MCP PyCon演示：连接LLM和现实世界API
   
完整演示 模型上下文协议（MCP） 展示了如何安全地将大型语言模型与现实世界的API连接起来。
🎯 挑战
在将LLM与内部API集成时，您将面临三个关键问题：
- 🌉 语言障碍LLM讲自然语言，API讲HTTP+JSON
- 🔒 安全困境：如何在不公开凭据的情况下授予LLM API访问权限
- 🎪 编排负担：业务逻辑在哪里？
这个演示使用MCP优雅地解决了这三个问题。
🏗️ 建筑
┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ Usuario │─────▶│ LLM │─────▶│ MCP Server │─────▶│ Task API │
│ (Español) │◀─────│ (Reasoning)│◀─────│ (Bridge) │◀─────│ (FastAPI) │
└─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘
│ │ │
Natural Language API Key (Secure) Stores in S3
Understanding Translation Layer JSON Files组件
- 任务API (快速API)：用于管理用户、调用和任务的RESTful API
- MCP服务器 (FastMCP)：使用现代Python装饰器将API公开为LLM工具的安全桥接
- 演示客户端 (溪流)：展示集成的交互式web应用程序
- S3存储：简单、持久的数据层（用于本地开发的LocalStack）
✨ 特性
- ✅ FastMCP框架：基于装饰器的MCP服务器，具有自动模式生成功能
- ✅ 安全凭据隔离：API密钥从未暴露于LLM
- ✅ 自然语言接口：西班牙语/英语命令→ API调用
- ✅ 多步骤编排：智能处理复杂的工作流程
- ✅ 完成CRUD操作：用户、预定通话和任务
- ✅ 生产准备就绪：FastAPI+AWS应用程序运行器+S3
- ✅ 互动演示：带有视觉场景的Streamlit web应用程序
- ✅ MCP检查员：交互式调试和测试工具
- ✅ Docker支持：完全集装箱化部署
🚀 快速开始
先决条件
- Python 3.12+
- Docker和Docker Compose（用于容器化设置）
- GitHub API密钥（用于LLM功能）
- 更多详情
选项1:Docker Compose（推荐）
# Clone repository
git clone https://github.com/ariosramirez/mcp-pycon.git
cd mcp-pycon
# Set GitHub API key for LLM features
export GITHUB_API_KEY=your-github-token-here
# Start all services
docker compose up -d
# View logs
docker compose logs -f服务地点：
- 任务API：http://localhost:8000 (文档)
- MCP服务器：http://localhost:8001
- 流媒体演示：http://localhost:8501
- 本地堆栈S3：http://localhost:4566
方案2：地方发展
# Install dependencies
uv sync
# OR
pip install -e .
# Configure environment
cp .env.example .env
# Edit .env with your credentials:
# - TASK_API_KEY
# - GITHUB_API_KEY
# - AWS configuration (for LocalStack: AWS_ENDPOINT_URL=http://localhost:4566)在单独的终端中启动服务：
# Terminal 1: Start Task API (with LocalStack)
cd task_api
docker compose up -d
# OR run directly
python -m task_api.main
# Terminal 2: Start MCP Server
python -m mcp_server.server
# Terminal 3: Start Streamlit Demo
streamlit run demo_client/streamlit_app.py🔍 MCP检查员
交互式测试和调试MCP工具：
# Make sure Task API is running
npx @modelcontextprotocol/inspector fastmcp run mcp_server/server.py:mcp打开http://localhost:5173致：
- 查看所有具有模式的可用工具
- 具有自定义参数的测试工具调用
- 检查请求/响应有效载荷
- 实时调试
示例：测试 register_user 工具
{
"name": "Test User",
"email": "test@example.com",
"company": "Test Corp"
}🎬 演示场景
Streamlit演示展示了三个场景，展示了MCP的功能：
场景1：注册和日程安排
*“请与联系人玛丽亚·加西亚注册我们的新客户‘Azollon International’。”(maria@test-azollon.com)并在本周五上午10点给他打电话。（二）*
展示：多步骤编排，安全的API调用
场景2：查询和更新
*“给我看看所有悬而未决的电话，并将第一个电话标记为已完成。”*
展示：数据检索、智能处理、更新
场景3：复杂的工作流程
*“为本周安排电话的所有客户创建跟踪任务。”*
展示：复杂的推理、数据聚合、编排
🚀 为什么选择FastMCP？
FastMCP通过以下方式减少样板 60% 与传统的MCP SDK相比：
传统MCP SDK：
@app.list_tools()
async def list_tools() -> list[Tool]:
return [Tool(name="...", inputSchema={...})] # Manual JSON schema
@app.call_tool()
async def call_tool(name: str, arguments: Any):
if name == "register_user": # Manual routing
return [TextContent(type="text", text="...")]FastMCP：
@mcp.tool
async def register_user(
name: Annotated[str, "Full name of the user"],
email: Annotated[str, "Email address"],
) -> str:
"""Register a new user."""
return "✅ User registered!" # Auto-wrapped!优点：
- 从类型提示自动生成架构
- 内置参数验证（Pydantic）
- 具有现代Python功能的类型安全
- 更简洁的错误处理
ToolError
📚 API概述
用户
POST /users-注册用户GET /users-列出所有用户GET /users/{user_id}-获取用户详细信息
预定通话
POST /calls-安排通话GET /calls-列出调用（可按user_id、status_filter过滤）PATCH /calls/{call_id}/status-更新状态
任务
POST /tasks-创建任务GET /tasks-列出任务（可按user_id、status_filter筛选）PATCH /tasks/{task_id}/status-更新状态
认证
所有端点都需要 X-API-Key 标题（除 /health):
curl -X POST http://localhost:8000/users \
-H "X-API-Key: demo-secret-key-change-in-production" \
-H "Content-Type: application/json" \
-d '{
"name": "María García",
"email": "maria@test-azollon.com",
"company": "Azollon International"
}'默认密钥： demo-secret-key-change-in-production
🧪 测试
快速健康检查
curl http://localhost:8000/health用Python测试
import httpx
client = httpx.Client(
base_url="http://localhost:8000",
headers={"X-API-Key": "demo-secret-key-change-in-production"}
)
# Create user
response = client.post("/users", json={
"name": "María García",
"email": "maria@test-azollon.com",
"company": "Azollon International"
})
user = response.json()
# Schedule call
response = client.post("/calls", json={
"user_id": user['id'],
"title": "Onboarding Call",
"scheduled_for": "2025-10-20T10:00:00Z",
"duration_minutes": 30
})🔒 安全最佳实践
- API密钥隔离：密钥仅存在于MCP服务器环境中，从不暴露于LLM
- AWS机密管理器：用于生产凭证管理
- HTTPS/TLS：始终在生产中启用
- 关键点旋转实施定期轮换政策
- 访问日志记录：监视所有API调用和工具使用
- 最小特权：授予最少的S3权限
🛠️ 项目结构
mcp-pycon-demo/
├── task_api/ # FastAPI application
│ ├── main.py # API endpoints
│ ├── models.py # Pydantic models
│ ├── storage.py # S3 storage layer
│ ├── Dockerfile # Container image
│ └── README.md # API documentation
├── mcp_server/ # MCP Server
│ ├── server.py # FastMCP implementation
│ └── README.md # MCP server documentation
├── demo_client/ # Streamlit Demo
│ ├── streamlit_app.py # Web UI
│ ├── langgraph_agent.py # LLM agent with LangGraph
│ └── azure_chat_wrapper.py # GitHub Models integration
├── docker-compose.yml # Container orchestration
├── pyproject.toml # Project metadata
└── README.md # This file➕ 添加新工具
1.添加API端点
@app.post("/your-endpoint")
async def your_endpoint(
data: YourModel,
api_key: str = Header(..., alias="X-API-Key")
):
verify_api_key(api_key)
return {"result": "success"}2.添加MCP工具
from typing import Annotated, Literal
from fastmcp.exceptions import ToolError
from pydantic import Field
@mcp.tool
async def your_new_tool(
param: Annotated[str, "Parameter description"],
count: Annotated[int, Field(ge=1, le=100)] = 10,
status: Annotated[Literal["active", "inactive"] | None, "Filter"] = None
) -> str:
"""Tool description for LLM."""
client = get_http_client()
try:
response = await client.post("/your-endpoint", json={"param": param})
response.raise_for_status()
return f"✅ Success: {response.json()}"
except httpx.HTTPStatusError as e:
raise ToolError(f"Failed: {e.response.json().get('message')}")FastMCP自动处理模式生成、验证和错误格式化！
🐳 Docker命令
# Start all services
docker compose up -d
# View logs (all)
docker compose logs -f
# View logs (specific service)
docker compose logs -f task-api
# Rebuild and restart
docker compose up -d --build
# Stop all services
docker compose down
# Stop and remove volumes
docker compose down -v💡 用例
这种架构非常适合：
- 内部工具集成：将LLM安全地连接到公司API
- 多服务编排：协调多个微服务
- Agent体系结构：构建自主的人工智能代理
- 企业人工智能：生产级LLM应用
- API民主化：API的自然语言访问
📖 资源
- FastMCP文档: https://gofastmcp.com
- MCP协议: https://modelcontextprotocol.io
- MCP检查员: https://github.com/modelcontextprotocol/inspector
- 快速API: https://fastapi.tiangolo.com
- LangGraph: https://langchain-ai.github.io/langgraph/
- GitHub模型: https://github.com/marketplace/models
🎯 与会者面临的挑战
想扩展演示吗？试试 PyCon挑战赛:
挑战.md -逐步添加指南：
- 📊 客户端摘要工具（数据聚合）
- 📝 用户信息提示（模板生成）
- 包括代码、测试说明和故障排除
时间： 30-40分钟| 困难： 中级
📁 文档
- 任务API自述 -API端点、测试、Docker设置
- MCP服务器自述文件 -MCP工具、FastMCP模式、检查器使用
- CLAUDE.md -克劳德代码开发指南
- 挑战.md -研讨会与会者的实践挑战
🤝 贡献
欢迎投稿！请随时提交拉取请求。
📝 许可证
此项目根据MIT许可证获得许可-有关详细信息，请参阅许可证文件。
🙏 致谢
- 为PyCon演示而构建
- 由FastMCP框架提供支持
- Anthropic的模型上下文协议
- web框架的FastAPI
- AWS用于无服务器基础架构
- 用于LLM访问的GitHub模型
______________________________________________________________________
让我们一起建设人工智能的未来！ 🚀