Saltearse al contenido

Configuración de Pydantic AI

Integre Pydantic AI con el servidor MCP de graph8. El modelo de agente tipado de Pydantic AI encaja a la perfección con graph8: los datos estructurados de sus prospectos fluyen a través de modelos Pydantic de extremo a extremo.

Requisitos previos

Python 3.10+
pydantic-ai con el extra MCP

pip install 'pydantic-ai[mcp]'

MCP alojado (OAuth remoto)

import asyncio
import os

from pydantic import BaseModel
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerHTTP


class Prospect(BaseModel):
    name: str
    company: str
    title: str
    linkedin_url: str | None = None


class ProspectList(BaseModel):
    prospects: list[Prospect]


server = MCPServerHTTP(url="https://be.graph8.com/mcp/")
agent = Agent(
    "anthropic:claude-sonnet-4",
    mcp_servers=[server],
    output_type=ProspectList,
    system_prompt=(
        "You are a SaaS prospector. Use g8_find_contacts to preview prospects. "
        "Never call credit-charging tools without explicit user confirmation."
    ),
)


async def main() -> None:
    async with agent.run_mcp_servers():
        result = await agent.run(
            "Find 10 VP Engineering at Series B SaaS startups in the US."
        )
        for p in result.output.prospects:
            print(f"{p.name} - {p.title} @ {p.company}")


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

La restricción output_type=ProspectList obliga al agente a devolver una lista tipada y validada, sin necesidad de análisis ni inferencias.

MCP autoalojado (stdio)

import os

from pydantic_ai.mcp import MCPServerStdio


server = MCPServerStdio(
    command="uvx",
    args=["g8-mcp-server"],
    env={
        "G8_API_KEY": os.environ["G8_API_KEY"],
        "G8_MCP_MODE": "gtm",
    },
)
# Same `agent = Agent(...)` as the hosted example.

Ejemplo práctico: flujo tipado de prospecto a secuencia

class SequenceEnrollment(BaseModel):
    sequence_name: str
    enrolled_count: int
    confirmation_required: bool


agent = Agent(
    "anthropic:claude-sonnet-4",
    mcp_servers=[server],
    output_type=SequenceEnrollment,
    system_prompt=(
        "Find prospects, save them to a list, and enroll in a sequence. "
        "Always set confirmation_required=true before calling "
        "g8_build_contact_list or g8_add_to_sequence."
    ),
)


async def main() -> None:
    async with agent.run_mcp_servers():
        result = await agent.run(
            "Enroll 25 VP Eng at Series B SaaS in the 'New SaaS Outreach' sequence."
        )
        if result.output.confirmation_required:
            input(f"Enroll {result.output.enrolled_count} contacts? [y/N] ")

Las salidas tipadas permiten condicionar con facilidad los pasos que consumen créditos a una lógica de confirmación explícita dentro de su código, en lugar de depender del modelo para mostrar un aviso.

Relacionado