Build a CrewAI News Research Crew with Real-Time Data (Python, 2025)

Why CrewAI + wpnews?

CrewAI agents are powerful but stateless — they have no knowledge of what happened in the news since their training cut-off. wpnews solves this by providing live AI and tech news through a clean Python API. The key fit: CrewAI's @tool decorator turns any Python function into an agent tool with zero boilerplate. You define the function, add a docstring, and CrewAI auto-generates the schema.

The get_morning_briefing() call is especially well-suited for CrewAI — it returns velocity status, breaking stories, and trending entities in a single call, giving the NewsResearcher agent structured situational awareness in one round-trip.

Step 1 — Install

pip install crewai wpnews

Get a free wpnews API key at wpnews.pro/api#get-key (1,000 calls/day free, no credit card).

Step 2 — Define wpnews tools

CrewAI's @tool decorator reads the docstring to build the agent's tool description. Keep docstrings precise — the LLM uses them to decide when to call each tool.

import os
from crewai.tools import tool
from wpnews import WPNews

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


@tool("Morning Briefing")
def get_morning_briefing(hours: int = 6) -> dict:
    """Get today's AI news situational awareness: velocity status (burst/normal/quiet),
    top breaking stories scored by freshness × quality, and trending entities.
    Call this first to understand the current news climate before searching further."""
    return _client.get_morning_briefing(hours=hours)


@tool("Search News")
def search_news(query: str) -> list:
    """Search recent AI and tech news articles by keyword or topic.
    Use for targeted searches after identifying topics from the morning briefing.
    Returns articles with title, summary, topics, entities, and URL."""
    return _client.search(q=query, limit=5)


@tool("Trending Entities")
def get_trending_entities(days: int = 7) -> list:
    """Get companies and people surging in AI news coverage over the past N days.
    Returns name, mention count, and velocity percentage. Use to identify
    which organizations are gaining or losing media attention."""
    return _client.get_trending_entities(days=days, limit=10)


@tool("Topic Digest")
def get_topic_digest(topic: str) -> dict:
    """Get a structured digest for a specific AI topic (e.g. 'large-language-models',
    'robotics', 'artificial-intelligence'). Returns article count, top articles,
    sentiment, and key entities for that topic."""
    return _client.get_agent_digest(lang="en", hours=24)

Step 3 — Create the agents

CrewAI agents are defined by role, goal, and backstory. These aren't just labels — the LLM uses them to maintain consistent behavior across the crew's conversation.

from crewai import Agent

news_researcher = Agent(
    role="AI News Researcher",
    goal=(
        "Gather comprehensive, real-time intelligence about the AI news landscape. "
        "Always start with the morning briefing to assess velocity, then investigate "
        "breaking stories and trending entities in depth."
    ),
    backstory=(
        "You are a veteran technology journalist with a nose for breaking news. "
        "You know that timing matters — a story from 2 hours ago is worth 10x "
        "a story from yesterday. You use data to decide what deserves attention."
    ),
    tools=[get_morning_briefing, search_news, get_trending_entities, get_topic_digest],
    verbose=True,
)

analyst = Agent(
    role="AI Industry Analyst",
    goal=(
        "Analyze raw news intelligence and identify the 2-3 most important business "
        "implications. Focus on what this means for companies building AI products."
    ),
    backstory=(
        "You are a former VC analyst who now advises AI startups. You translate "
        "news into strategic signals: funding shifts, competitive moves, regulatory "
        "risk, and technology adoption curves."
    ),
    tools=[],  # analyst reasons from researcher's output, no tools needed
    verbose=True,
)

writer = Agent(
    role="Technical Content Writer",
    goal=(
        "Transform research and analysis into a polished, scannable daily briefing "
        "that busy engineers can read in 90 seconds. Use bullet points and headers."
    ),
    backstory=(
        "You write for AI practitioners, not executives. Your briefings are concrete, "
        "specific, and skippable — readers can jump to what matters to them."
    ),
    tools=[],
    verbose=True,
)

Step 4 — Define tasks and run the crew

from crewai import Task, Crew, Process

research_task = Task(
    description=(
        "Run a full morning briefing using get_morning_briefing(). "
        "If velocity is 'burst', search for the specific breaking stories. "
        "Compile: velocity status, top 3 breaking stories with scores, "
        "top 5 trending entities with velocity percentages."
    ),
    expected_output=(
        "Structured JSON with: velocity_status, hot_articles (list of 3), "
        "trending_entities (list of 5), key_topics (list of topics that appear most)."
    ),
    agent=news_researcher,
)

analysis_task = Task(
    description=(
        "Receive the research output and identify the top 2-3 business implications "
        "for AI product teams. Each implication should include: what happened, "
        "why it matters, and what to watch next."
    ),
    expected_output=(
        "3 bullet points. Each: one-line headline + 2-sentence explanation. "
        "No fluff. No 'it remains to be seen'."
    ),
    agent=analyst,
    context=[research_task],
)

write_task = Task(
    description=(
        "Write a daily AI news briefing combining the research and analysis. "
        "Format: ## Velocity + breaking stories section, ## Business implications, "
        "## Trending entities table. Under 300 words total."
    ),
    expected_output=(
        "A markdown briefing with 3 sections. Ready to post to Slack or send by email."
    ),
    agent=writer,
    context=[research_task, analysis_task],
)

crew = Crew(
    agents=[news_researcher, analyst, writer],
    tasks=[research_task, analysis_task, write_task],
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff()
print(result.raw)

Full working script (under 120 lines)

"""
CrewAI + wpnews daily briefing crew
Run: OPENAI_API_KEY=... WPNEWS_API_KEY=... python crewai_news.py
"""
import os
from crewai import Agent, Task, Crew, Process
from crewai.tools import tool
from wpnews import WPNews

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

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

@tool("Search News")
def search_news(query: str) -> list:
    """Search recent AI and tech news by keyword. Returns 5 articles with summaries."""
    return _news.search(q=query, limit=5)

@tool("Trending Entities")
def get_trending_entities(days: int = 7) -> list:
    """Companies and people trending in AI news — name, count, velocity percentage."""
    return _news.get_trending_entities(days=days, limit=10)

researcher = Agent(
    role="AI News Researcher",
    goal="Gather real-time AI news intelligence. Start with morning briefing.",
    backstory="Veteran tech journalist — timing matters, data drives decisions.",
    tools=[get_morning_briefing, search_news, get_trending_entities],
    verbose=True,
)

analyst = Agent(
    role="AI Industry Analyst",
    goal="Identify 2-3 business implications for AI product teams.",
    backstory="Ex-VC analyst translating news into strategic signals.",
    tools=[],
    verbose=True,
)

writer = Agent(
    role="Technical Content Writer",
    goal="Write a scannable 90-second briefing. Concrete, specific, no fluff.",
    backstory="Writes for AI practitioners — bullet points, headers, skippable.",
    tools=[],
    verbose=True,
)

research = Task(
    description="Run morning briefing. If burst, search for breaking stories. "
                "Return: velocity_status, top 3 hot articles, top 5 entities.",
    expected_output="JSON with velocity_status, hot_articles[3], trending_entities[5].",
    agent=researcher,
)

analysis = Task(
    description="Top 2-3 business implications from the research. What happened, "
                "why it matters, what to watch next.",
    expected_output="3 bullet points — headline + 2-sentence explanation each.",
    agent=analyst,
    context=[research],
)

writing = Task(
    description="Combine research + analysis into a markdown briefing: velocity section, "
                "business implications, trending entities. Under 300 words.",
    expected_output="Markdown briefing ready for Slack or email.",
    agent=writer,
    context=[research, analysis],
)

crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research, analysis, writing],
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff()
print(result.raw)

Hierarchical crews (manager pattern)

For more autonomous operation, switch to Process.hierarchical. This adds a manager LLM that coordinates agents and decides task sequencing:

from crewai import LLM

crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research, analysis, writing],
    process=Process.hierarchical,
    manager_llm=LLM(model="gpt-4o"),
    verbose=True,
)

In hierarchical mode, the manager agent decides which agent to call and in what order — useful when task sequencing depends on content (e.g., skip analysis if no breaking news).

Use all 69 tools automatically

Instead of defining tools manually, fetch all 69 pre-built OpenAI schemas and wrap them as CrewAI tools:

import requests
from crewai.tools import BaseTool
from pydantic import BaseModel

# Fetch all 69 schemas from API
all_tools_raw = requests.get(
    "https://api.wpnews.pro/api/v1/openai-tools.json",
    headers={"X-API-Key": os.environ.get("WPNEWS_API_KEY", "")}
).json()

# Dynamic CrewAI tool wrapper
class WPNewsTool(BaseTool):
    name: str
    description: str
    _endpoint_name: str

    def __init__(self, schema: dict):
        fn = schema["function"]
        super().__init__(name=fn["name"], description=fn["description"])
        self._endpoint_name = fn["name"]

    def _run(self, **kwargs) -> dict:
        resp = requests.post(
            f"https://api.wpnews.pro/api/v1/openai-tools/{self._endpoint_name}/call",
            json=kwargs,
            headers={"X-API-Key": os.environ.get("WPNEWS_API_KEY", "")}
        )
        return resp.json()

# All 69 tools available to any agent
ALL_WPNEWS_TOOLS = [WPNewsTool(t) for t in all_tools_raw]

Scheduling the crew

# crontab -e
0 8 * * 1-5 OPENAI_API_KEY=... WPNEWS_API_KEY=... python /path/to/crewai_news.py >> ~/briefings.log 2>&1

Or use Prefect, Airflow, or GitHub Actions for richer orchestration and retry logic.