Build a News Agent with Pydantic AI + wpnews (Python, 2025)

Why Pydantic AI + wpnews?

Pydantic AI is the type-safe agent framework from Samuel Colvin (creator of Pydantic). Unlike LangChain or CrewAI, every tool return value is a validated Pydantic model — no JSON parsing, no KeyError, no silent type coercion. The agent's result is also typed, so your IDE and type checker know exactly what the agent returns.

wpnews pairs naturally with this approach because our Python SDK returns typed dicts with consistent schemas. Wrap them in Pydantic models and you get an end-to-end type-safe news intelligence pipeline.

What you'll build

• A Pydantic AI agent with wpnews tools (morning briefing, search, entities)
• Typed result models — NewsBriefing, BreakingAlert — validated by Pydantic
• A streaming agent that yields results as they come

Step 1 — Install

pip install pydantic-ai wpnews

Get a free wpnews API key at wpnews.pro/api#get-key.

Step 2 — Define typed result models

The agent's output is a validated Pydantic model. Define what you want the agent to produce — Pydantic AI enforces the schema.

from pydantic import BaseModel, Field
from typing import Literal

class HotArticle(BaseModel):
    title: str
    hot_score: float
    hours_ago: float
    topics: list[str] = []

class TrendingEntity(BaseModel):
    name: str
    velocity_pct: float = 0.0
    recent_count: int = 0

class NewsBriefing(BaseModel):
    velocity_status: Literal["burst", "normal", "quiet"]
    summary: str = Field(description="One-paragraph situation summary")
    hot_articles: list[HotArticle]
    trending_entities: list[TrendingEntity]
    action_items: list[str] = Field(
        description="2-3 specific actions an AI product team should take today"
    )

Step 3 — Define wpnews tools

Pydantic AI tools are plain functions with type annotations. The framework generates the JSON schema automatically — no decorators or schema dicts needed.

import os
from wpnews import WPNews

_client = WPNews(api_key=os.environ.get("WPNEWS_API_KEY", ""))


def get_morning_briefing(hours: int = 6) -> dict:
    """Get today's AI news situational awareness.

    Returns velocity_status (burst/normal/quiet), hot breaking stories ranked
    by freshness × quality, and trending entities. Call this first to understand
    the current news climate before searching further.

    Args:
        hours: Look-back window for breaking stories (1-24 hours, default 6).
    """
    return _client.get_morning_briefing(hours=hours)


def search_news(query: str, limit: int = 5) -> list[dict]:
    """Search recent AI and tech news articles by keyword.

    Use after get_morning_briefing() for targeted investigation of specific topics
    or entities that appear in the briefing.

    Args:
        query: Search query (e.g. 'GPT-5 training', 'Anthropic funding').
        limit: Number of results (1-20, default 5).
    """
    return _client.search(q=query, limit=limit)


def get_trending_entities(days: int = 7, limit: int = 10) -> list[dict]:
    """Get companies and people trending in AI news coverage.

    Returns entities ranked by velocity — how much their mention count has
    accelerated relative to their baseline. Useful for spotting emerging players.

    Args:
        days: Look-back window (1-30, default 7).
        limit: Number of entities to return (default 10).
    """
    return _client.get_trending_entities(days=days, limit=limit)

Step 4 — Create the agent

from pydantic_ai import Agent

news_agent = Agent(
    "openai:gpt-4o",
    output_type=NewsBriefing,
    tools=[get_morning_briefing, search_news, get_trending_entities],
    system_prompt=(
        "You are an AI news analyst. When asked for a briefing:\n"
        "1. Call get_morning_briefing() first to assess the news climate.\n"
        "2. If velocity_status is 'burst', search for the top breaking stories.\n"
        "3. Return a structured NewsBriefing with actionable insights.\n"
        "Focus on implications for AI product teams."
    ),
)

Step 5 — Run and get typed results

import asyncio

async def main():
    result = await news_agent.run(
        "Give me today's AI news briefing with action items."
    )

    # result.output is a validated NewsBriefing — fully typed
    briefing: NewsBriefing = result.output

    print(f"Velocity: {briefing.velocity_status.upper()}")
    print(f"\nSummary: {briefing.summary}")

    print(f"\nBreaking stories ({len(briefing.hot_articles)}):")
    for article in briefing.hot_articles:
        print(f"  🔥 {article.hot_score:.1f}pt [{article.hours_ago:.1f}h] {article.title}")

    print(f"\nTrending entities:")
    for entity in briefing.trending_entities[:5]:
        print(f"  • {entity.name} ({entity.velocity_pct:.1f}% velocity)")

    print(f"\nAction items:")
    for item in briefing.action_items:
        print(f"  → {item}")

asyncio.run(main())

Streaming results

Pydantic AI supports streaming — useful when the agent is generating a long report and you want to show progress:

async def stream_briefing():
    async with news_agent.run_stream(
        "Detailed AI news briefing for our engineering team."
    ) as stream:
        async for chunk in stream.stream_text():
            print(chunk, end="", flush=True)
    # Final validated result after streaming completes
    briefing = await stream.get_output()
    return briefing

Full working script (under 80 lines)

"""
Pydantic AI + wpnews news analyst agent
Run: OPENAI_API_KEY=... WPNEWS_API_KEY=... python pydantic_ai_news.py
"""
import os, asyncio
from typing import Literal
from pydantic import BaseModel, Field
from pydantic_ai import Agent
from wpnews import WPNews

_news = WPNews(api_key=os.environ.get("WPNEWS_API_KEY", ""))


class HotArticle(BaseModel):
    title: str
    hot_score: float
    hours_ago: float
    topics: list[str] = []

class TrendingEntity(BaseModel):
    name: str
    velocity_pct: float = 0.0
    recent_count: int = 0

class NewsBriefing(BaseModel):
    velocity_status: Literal["burst", "normal", "quiet"]
    summary: str = Field(description="One-paragraph situation summary")
    hot_articles: list[HotArticle]
    trending_entities: list[TrendingEntity]
    action_items: list[str] = Field(description="2-3 actionable items for AI product teams")


def get_morning_briefing(hours: int = 6) -> dict:
    """Velocity + breaking stories + trending entities. Call this first."""
    return _news.get_morning_briefing(hours=hours)

def search_news(query: str, limit: int = 5) -> list[dict]:
    """Search recent AI/tech news by keyword. Use for follow-up investigation."""
    return _news.search(q=query, limit=limit)

def get_trending_entities(days: int = 7, limit: int = 10) -> list[dict]:
    """Companies and people trending in AI coverage — name, velocity, mention count."""
    return _news.get_trending_entities(days=days, limit=limit)


agent = Agent(
    "openai:gpt-4o",
    output_type=NewsBriefing,
    tools=[get_morning_briefing, search_news, get_trending_entities],
    system_prompt=(
        "You are an AI news analyst. Call get_morning_briefing() first. "
        "If burst, search for breaking stories. Return structured NewsBriefing."
    ),
)


async def main():
    result = await agent.run("Today's AI news briefing with action items.")
    b = result.output
    print(f"[{b.velocity_status.upper()}] {b.summary}\n")
    print("Breaking stories:")
    for a in b.hot_articles:
        print(f"  🔥 {a.hot_score:.1f}pt [{a.hours_ago:.1f}h] {a.title}")
    print("\nAction items:")
    for item in b.action_items:
        print(f"  → {item}")

asyncio.run(main())

Why typed outputs matter

When your agent returns a validated NewsBriefing model, downstream code is safe:

# Type-safe downstream: IDE knows briefing.hot_articles is list[HotArticle]
if briefing.velocity_status == "burst":
    for article in briefing.hot_articles[:3]:
        send_slack_alert(article.title, article.hot_score)

# No KeyError, no None checks, no JSON parsing
top_entity = briefing.trending_entities[0]
track_entity(top_entity.name, top_entity.velocity_pct)

Compare to untyped agent outputs: result["trending_entities"][0]["velocity_pct"] — no IDE support, runtime KeyErrors, schema drift. Pydantic AI eliminates this class of bugs entirely.