Implementation:PrefectHQ Prefect AI Cleanup Agent

Metadata
Sources	Prefect, pydantic-ai MCP
Domains	AI_Agents, Orchestration
Last Updated	2026-02-09 00:00 GMT

Overview

Concrete wrapper for creating an AI approval agent with Prefect MCP tools for autonomous workflow decisions.

Description

The create_cleanup_agent function creates a PrefectAgent configured with an Anthropic Claude model, MCP server connection to Prefect tools, and structured output (CleanupDecision). The MCP server provides read-only tools for investigating Prefect instance health. This is a Wrapper Doc combining pydantic-ai, pydantic-ai-mcp, and Prefect.

Code Reference

• Repository: https://github.com/PrefectHQ/prefect
• File: examples/ai_database_cleanup_with_approval.py (L138-187) for agent creation and approval flow
• Signature:

def create_cleanup_agent() -> PrefectAgent[None, CleanupDecision]:
    mcp_server = MCPServerStdio(
        "prefect", "uvx", args=["--from", "prefect-mcp", "prefect-mcp-server"]
    )
    agent = Agent(
        model=AnthropicModel("claude-sonnet-4-5-20250929"),
        output_type=CleanupDecision,
        system_prompt=AGENT_PROMPT,
        mcp_servers=[mcp_server],
    )
    return PrefectAgent(
        agent,
        model_task_config=TaskConfig(retries=2, timeout_seconds=120.0),
    )

• Import: from pydantic_ai import Agent; from pydantic_ai.models.anthropic import AnthropicModel; from pydantic_ai.durable_exec.prefect import PrefectAgent, TaskConfig; from pydantic_ai_mcp import MCPServerStdio

I/O Contract

Inputs

• context (str) — description of proposed cleanup including retention period, states, count, preview

Outputs

• CleanupDecision — structured decision with approved (bool), confidence (float), reasoning (str), concerns (list[str])

Usage Example

from pydantic import BaseModel, Field
from pydantic_ai import Agent
from pydantic_ai.models.anthropic import AnthropicModel
from pydantic_ai.durable_exec.prefect import PrefectAgent, TaskConfig
from pydantic_ai_mcp import MCPServerStdio

class CleanupDecision(BaseModel):
    approved: bool
    confidence: float = Field(ge=0.0, le=1.0)
    reasoning: str
    concerns: list[str] | None = None

def create_cleanup_agent():
    mcp_server = MCPServerStdio(
        "prefect", "uvx", args=["--from", "prefect-mcp", "prefect-mcp-server"]
    )
    agent = Agent(
        model=AnthropicModel("claude-sonnet-4-5-20250929"),
        output_type=CleanupDecision,
        system_prompt="You are an operations agent reviewing database cleanup...",
        mcp_servers=[mcp_server],
    )
    return PrefectAgent(agent, model_task_config=TaskConfig(retries=2, timeout_seconds=120.0))

Related Pages

• Principle:PrefectHQ_Prefect_AI_Approval_Gate
• Environment:PrefectHQ_Prefect_AI_Integration_Credentials