Type-Safe Web Scraping for Pydantic AI Agents

pip install pydantic-ai requests

from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext
import requests

class ScrapedPage(BaseModel):
    """Structured result from a web scrape."""
    url: str
    title: str = ""
    content: str = ""
    status_code: int = 200

class ProductInfo(BaseModel):
    """Extracted product data."""
    name: str
    price: float
    currency: str = "USD"
    in_stock: bool = True
    description: str = ""

# Create the agent with a result type
agent = Agent(
    "openai:gpt-4",
    system_prompt="You are a web research assistant. Use the scrape tool to gather information.",
)

MANTIS_API_KEY = "YOUR_API_KEY"

@agent.tool
async def scrape_webpage(ctx: RunContext, url: str) -> ScrapedPage:
    """Scrape a webpage and return its content."""
    resp = requests.post(
        "https://api.mantisapi.com/v1/scrape",
        headers={"x-api-key": MANTIS_API_KEY},
        json={"url": url, "render_js": True}
    )
    data = resp.json()
    return ScrapedPage(
        url=url,
        title=data.get("title", ""),
        content=data.get("content", ""),
        status_code=resp.status_code
    )

@agent.tool
async def extract_product(ctx: RunContext, url: str) -> ProductInfo:
    """Extract structured product data from a URL using AI."""
    resp = requests.post(
        "https://api.mantisapi.com/v1/extract",
        headers={"x-api-key": MANTIS_API_KEY},
        json={
            "url": url,
            "schema": {
                "name": "string",
                "price": "number",
                "currency": "string",
                "in_stock": "boolean",
                "description": "string"
            }
        }
    )
    data = resp.json().get("extracted", {})
    return ProductInfo(**data)

import asyncio

async def main():
    result = await agent.run(
        "Find the price of the MacBook Pro M4 on Apple's website"
    )
    print(result.data)

asyncio.run(main())