Durable Execution MCP Demo

PydanticAI+预执行代理+预执行MCP演示

这表明了什么

• 持久执行：模型请求和MCP工具调用被跟踪为具有自动重试功能的Prefect任务
• 精确的本地配置：使用Prefect变量和秘密块，而不是环境变量
• 代码存储器：使用从GitHub仓库部署 flow.from_source()
• MCP集成：连接到Prefect MCP服务器以查询和管理Prefect资源
• 管理执行：没有使用Prefect的托管工作池进行管理的基础设施
• 可选观察性：Logfire集成用于详细跟踪

先决条件

1. Prefect Cloud帐户 - 注册
2. API型号键 -选择一个：

- 无烟煤API键 （克劳德） - OpenAI API密钥 GPT

1. Python 3.10+

设置

1.创建精确变量

# MCP server URL (required)
prefect variable set fastmcp-server-url "https://your-mcp-server-url/mcp"

# Model to use (optional, defaults to Claude 3.5 Sonnet)
prefect variable set pydantic-ai-model "anthropic:claude-3-5-sonnet-20241022"

2.创建完美的秘密块

# Register the secret block type
prefect block register -m prefect.blocks.system

然后使用Python创建secrets：

from prefect.blocks.system import Secret

# For FastMCP server authentication (required)
fastmcp_secret = Secret(value="your-fastmcp-auth-token")
fastmcp_secret.save(name="fastmcp-auth-token")

# For Anthropic Claude models (required if using Claude)
anthropic_secret = Secret(value="your-anthropic-api-key")
anthropic_secret.save(name="anthropic-api-key")

# For OpenAI models (required if using OpenAI)
openai_secret = Secret(value="your-openai-api-key")
openai_secret.save(name="openai-api-key")

# For Logfire observability (optional)
logfire_secret = Secret(value="your-logfire-token")
logfire_secret.save(name="logfire-token")

3.部署到Prefect Cloud

cd examples/pydantic-ai-demo
python deploy.py

这将：

• 使用从GitHub拉取代码 flow.from_source()
• 从Prefect变量和机密加载配置
• 使用Prefect托管执行创建部署
• 不需要工人！

用法

触发跑步

# Run with default prompt (dashboard overview)
prefect deployment run 'pydantic-ai-mcp-demo/pydantic-ai-mcp-demo'

# Run with a custom prompt
prefect deployment run 'pydantic-ai-mcp-demo/pydantic-ai-mcp-demo' \
  --param prompt='List my most recent failing flow runs'

# Run with a different model
prefect deployment run 'pydantic-ai-mcp-demo/pydantic-ai-mcp-demo' \
  --param model='openai:gpt-4o'

示例提示

# Dashboard overview
--param prompt='Show me a dashboard overview of my Prefect instance'

# Debug failing runs
--param prompt='List my most recent failing flow runs and tell me what failed'

# Check deployments
--param prompt='Show me all my active deployments'

# Work pool status
--param prompt='Show me the status of my work pools including active workers'

查看结果

完美的云用户界面：在以下位置查看流执行、任务分解和日志https://app.prefect.cloud

您将看到：

• create-prefect-mcp-agent 任务（使用API密钥加载创建代理）
• prefect-assistant_model_request 任务（LLM调用）
• prefect-assistant_mcp_[tool_name] 任务（MCP工具调用）

Logfire用户界面 （如果已配置）：查看代理执行、模型调用和工具使用的详细跟踪

配置

变量

• fastmcp服务器url （必填）：您的MCP服务器的URL
• pydantic ai模型 （可选）：要使用的模型（默认为 anthropic:claude-3-5-sonnet-20241022)

秘密方块

• fastmcp身份验证令牌 （必需）：FastMCP服务器的身份验证令牌
• 人工api密钥 （Claude必需）：Anthropic API密钥
• openai api密钥 （OpenAI必需）：OpenAI API密钥
• logfire令牌 （可选）：用于可观察性的Logfire令牌

流根据模型提供程序（anthropic:或openai:）自动加载相应的API密钥。

支持的型号

人类学（克劳德）

• anthropic:claude-3-5-sonnet-20241022 （默认，推荐）
• anthropic:claude-3-opus-20240229
• anthropic:claude-3-sonnet-20240229

OpenAI（GPT）

• openai:gpt-4o
• openai:gpt-4o-mini
• openai:gpt-4-turbo

本地测试（可选）

对于本地开发，您可以设置环境变量：

export FASTMCP_SERVER_URL='https://your-server.fastmcp.app/mcp'
export ANTHROPIC_API_KEY='sk-ant-...'

python demo_flow.py

注意：部署时，流从Prefect变量和secrets加载配置，而不是环境变量。

建筑

┌─────────────────────┐
│  Prefect Cloud      │
│  (Managed Execution)│
│                     │
│  ┌───────────────┐  │
│  │ PydanticAI    │  │
│  │ Agent Flow    │  │
│  │ + PrefectAgent│  │
│  └───────┬───────┘  │
└──────────┼──────────┘
           │
           │ MCP Protocol
           │
           ▼
┌─────────────────────┐
│  Prefect MCP Server │
│                     │
│  Tools:             │
│  - get_dashboard    │
│  - get_flow_runs    │
│  - get_deployments  │
│  - get_work_pools   │
│  - etc.             │
└─────────────────────┘
           │
           │ Prefect API
           │
           ▼
┌─────────────────────┐
│  Prefect Cloud API  │
└─────────────────────┘

主要特点

1. 持久执行：每个模型请求和工具调用都是一个跟踪的Prefect任务，会自动重试
2. 本机配置：使用Prefect变量和机密进行安全配置管理
3. 代码存储器：从GitHub部署，无需本地代码打包
4. 自动密钥加载：流基于模型提供程序加载正确的API密钥
5. 可选观察性：添加Logfire令牌以进行详细跟踪
6. 托管基础设施：无需运行工人或建造容器

资源

• PydanticAI文档
• 完美代理指南
• 完善的文件
• 完善MCP服务器
• Logfire文件