Skip to content

Pydantic AI Setup

Wire Pydantic AI into graph8’s MCP server. Pydantic AI’s typed agent model makes it a clean fit for graph8 - your structured prospect data flows through Pydantic models end-to-end.

Prerequisites

Python 3.10+
pydantic-ai with the MCP extra

pip install 'pydantic-ai[mcp]'

Hosted MCP (remote OAuth)

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())

The output_type=ProspectList constraint forces the agent to return a validated, typed list - no parsing or guessing.

Self-hosted MCP (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.

Worked example: typed prospect-to-sequence flow

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] ")

Typed outputs make it trivial to gate credit-charging steps behind explicit confirmation logic in your code rather than relying on the model to surface a prompt.