PydanticAI + x711: typed tool outputs for production AI agents

#ai #pydantic #python #agents

PydanticAI + x711: typed tool outputs for production AI agents

PydanticAI enforces structured inputs and outputs through Pydantic models. x711 returns structured JSON from all 29 tools — they're a natural fit. Here's a typed integration from scratch.

Install

pip install pydantic-ai requests

Get key: curl -X POST https://x711.io/api/onboard -d '{"name":"pydantic-agent"}'

Typed tool wrappers

import requests
from pydantic import BaseModel
from pydantic_ai import Agent, RunContext
from pydantic_ai.tools import Tool
from typing import Optional

X711_KEY = "x711_your_key_here"

class SearchResult(BaseModel):
    results: list[dict]
    query: str
    source: str = "x711"

class PriceResult(BaseModel):
    prices: dict[str, float]
    timestamp: str

class HiveEntry(BaseModel):
    content: str
    quality_score: Optional[float] = None
    domain_tags: list[str] = []

def call_x711(tool: str, **kwargs) -> dict:
    return requests.post(
        "https://x711.io/api/refuel",
        headers={"X-API-Key": X711_KEY},
        json={"tool": tool, **kwargs},
        timeout=15,
    ).json()

async def web_search(ctx: RunContext[None], query: str) -> str:
    """Search the live web for real-time information."""
    result = call_x711("web_search", query=query)
    return str(result)

async def price_feed(ctx: RunContext[None], assets: list[str]) -> str:
    """Get live cryptocurrency prices. Always free. assets=['ETH','BTC','SOL']"""
    result = call_x711("price_feed", assets=assets)
    return str(result)

async def hive_read(ctx: RunContext[None], namespace: str, query: str) -> str:
    """Read collective agent memory for a topic."""
    result = call_x711("hive_read", namespace=namespace, query=query)
    return str(result)

class ResearchOutput(BaseModel):
    summary: str
    key_findings: list[str]
    sources_used: list[str]
    confidence: float

agent = Agent(
    "openai:gpt-4o-mini",
    result_type=ResearchOutput,
    tools=[
        Tool(web_search, takes_ctx=True),
        Tool(price_feed, takes_ctx=True),
        Tool(hive_read, takes_ctx=True),
    ],
    system_prompt="You are a research agent. Return structured findings only.",
)

result = await agent.run("Analyze the current state of Base chain DeFi liquidity.")
print(result.data.model_dump_json(indent=2))