Add live web search to a Pydantic AI agent

Pydantic AI supports MCP natively — load Superhighway's five search tools via MCPServerStdio in a few lines, or define typed @agent.tool_plain functions that call the REST API with a free key.

Two paths

Path 1: MCP server (recommended for autonomous agents)

pip install pydantic-ai httpx

import asyncio
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStdio

server = MCPServerStdio(
    "npx",
    args=["-y", "superhighway-mcp"],
    env={
        "AGENT_PRIVATE_KEY": "0xYOUR_FUNDED_BASE_WALLET_KEY",
        "X402_NETWORK": "base",
    },
)

agent = Agent("openai:gpt-4o", mcp_servers=[server])

async def main():
    async with agent.run_mcp_servers():
        result = await agent.run("What are the latest developments in AI agent frameworks?")
        print(result.data)

asyncio.run(main())

The agent gets all five tools — web_search, news_search, image_search, scrape, research — and each call settles a $0.001–$0.005 USDC payment automatically. See wallet setup or the x402 flow.

Path 2: @agent.tool_plain with an API key

Get a free API key (1,000 calls/month) and define typed tools directly on the agent:

import asyncio, os
import httpx
from pydantic_ai import Agent

agent = Agent(
    "openai:gpt-4o",
    system_prompt="Use web_search, news_search, and research tools to answer with current information.",
)

@agent.tool_plain
def web_search(query: str, limit: int = 5) -> str:
    """Search the live web for current events, facts, and research."""
    r = httpx.get("https://superhighway.walls.sh/search",
        params={"q": query, "limit": limit},
        headers={"Authorization": f"Bearer {os.environ['SUPERHIGHWAY_API_KEY']}"},
        timeout=10)
    r.raise_for_status()
    results = r.json()["results"]
    return "\n\n".join(f"**{x['title']}**\n{x['url']}\n{x['description']}" for x in results)

@agent.tool_plain
def news_search(query: str) -> str:
    """Search recent news articles with published dates."""
    r = httpx.get("https://superhighway.walls.sh/news",
        params={"q": query},
        headers={"Authorization": f"Bearer {os.environ['SUPERHIGHWAY_API_KEY']}"},
        timeout=10)
    r.raise_for_status()
    results = r.json()["results"]
    return "\n\n".join(
        f"**{x['title']}** ({x.get('publishedDate', '')})\n{x['url']}\n{x['description']}"
        for x in results)

@agent.tool_plain
def research(query: str) -> str:
    """Search + read the top matching pages for a query — returns full content."""
    r = httpx.get("https://superhighway.walls.sh/research",
        params={"q": query, "pages": 2},
        headers={"Authorization": f"Bearer {os.environ['SUPERHIGHWAY_API_KEY']}"},
        timeout=30)
    r.raise_for_status()
    data = r.json()
    return "\n\n---\n\n".join(
        f"## {p['title']}\n{p['url']}\n\n{p['markdown'][:3000]}"
        for p in data["pages"] if not p.get("error"))

async def main():
    result = await agent.run("Summarize the top AI news from the last 48 hours.")
    print(result.data)

asyncio.run(main())

@agent.tool_plain is for tools that don't need access to the run context. Use @agent.tool (with a RunContext first argument) if you want access to deps or model settings inside the tool body.

Which tools are available

Full API reference: /openapi.json. Get a free API key or learn the x402 wallet flow.