Skip to content
unlob

CrewAI · Python

Build a research agent with CrewAI

Build a research crew on CrewAI, giving a researcher and a fact-checker separate roles over the same unlob index.

What you are building

The crew

Research analyst

Gathers with sources

Fact checker

Re-searches at min_sources=3

Shared tool

UnlobSearch

One tool, both agents

Services

Anthropic API

claude-opus-5

unlob

Your own index

Your side of the API. unlob is one HTTPS service among the others — nothing here is specific to how our index works, so this agent survives a change of framework.

Install

Dependenciesbash
pip install crewai crewai-tools httpx

Already have CrewAI wired up? TheCrewAI integration pagehas the connection config on its own — this page assumes you have it and gets on with building.

Step by step

  1. Write the tool

    CrewAI tools subclass BaseTool with a Pydantic args schema. The description is what every agent in the crew reads.

    tools.pypython
    import os, httpx
    from typing import Type
    from crewai.tools import BaseTool
    from pydantic import BaseModel, Field
    
    
    class SearchInput(BaseModel):
        query: str = Field(description="What to search for")
        min_sources: int = Field(default=2, description="Independent sources required")
    
    
    class UnlobSearch(BaseTool):
        name: str = "web_search"
        description: str = (
            "Search the web on unlob's own index. Returns metadata only — url, "
            "title, snippet and independent_sources. Never returns page bodies."
        )
        args_schema: Type[BaseModel] = SearchInput
    
        def _run(self, query: str, min_sources: int = 2) -> str:
            r = httpx.get("https://api.unlob.com/search",
                          headers={"x-api-key": os.environ["UNLOB_API_KEY"]},
                          params={"q": query,
                                  "min_independent_sources": min_sources,
                                  "limit": 8}, timeout=15)
            r.raise_for_status()
            return "\n".join(
                f"[{h['independent_sources']} sources] {h['title']} — {h['url']}\n  {h['snippet']}"
                for h in r.json()["results"]
            )
  2. Give the crew two roles

    A researcher gathers and a fact-checker challenges. The second agent is the point — it re-queries with a higher corroboration floor rather than trusting the first.

    crew.pypython
    from crewai import Agent, Task, Crew, Process
    
    search = UnlobSearch()
    
    researcher = Agent(
        role="Research analyst",
        goal="Find what is actually known about {topic}, with sources",
        backstory="You search before you assert, and you keep every URL.",
        tools=[search],
        llm="claude-opus-5",
    )
    
    checker = Agent(
        role="Fact checker",
        goal="Find claims that rest on a single source and flag them",
        backstory=(
            "You are sceptical by default. You re-search each claim with "
            "min_sources=3 and report what fails to corroborate."
        ),
        tools=[search],
        llm="claude-opus-5",
    )
  3. Wire the tasks so the checker sees the findings

    This is the step that makes the crew worth having. Passing the first task in the second task’s context is what hands the researcher’s output to the fact-checker — omit it and the checker verifies nothing while appearing to work.

    tasks.pypython
    from crewai import Task, Crew, Process
    
    gather = Task(
        description="Research {topic}. Return findings, each with its URL.",
        expected_output="A bulleted list of findings, each ending with its source URL.",
        agent=researcher,
    )
    
    verify = Task(
        description=(
            "Take the findings and re-search each with min_sources=3. Mark any "
            "that fail to corroborate as UNVERIFIED and explain why."
        ),
        expected_output="The findings, each marked VERIFIED or UNVERIFIED with a reason.",
        agent=checker,
        context=[gather],   # <- without this the checker never sees the findings
    )
    
    crew = Crew(
        agents=[researcher, checker],
        tasks=[gather, verify],
        process=Process.sequential,
    )
    
    result = crew.kickoff(inputs={"topic": "EU AI Act GPAI rule changes in 2026"})

What happens when it runs

YouResearcherFact checkerunlobkickoff(topic)web_search(min_sources=2)hitsfindings (task context)web_search(min_sources=3)fewer, better-supported hitsverified / unverified
One turn of the loop. Search returns metadata only, so the agent spends a few hundred tokens deciding rather than tens of thousands reading.

The whole thing

Complete and runnable. Set the two environment variables and it works.

research_crew.py — completepython
"""A two-role research crew on CrewAI, backed by the unlob web index.

    pip install crewai crewai-tools httpx
    export UNLOB_API_KEY=ulb_...
    export ANTHROPIC_API_KEY=sk-ant-...
"""
import os
from typing import Type

import httpx
from crewai import Agent, Task, Crew, Process
from crewai.tools import BaseTool
from pydantic import BaseModel, Field


class SearchInput(BaseModel):
    query: str = Field(description="What to search for")
    min_sources: int = Field(default=2, description="Independent sources required")


class UnlobSearch(BaseTool):
    name: str = "web_search"
    description: str = (
        "Search the web on unlob's own index. Returns metadata only — url, "
        "title, snippet and independent_sources. Never returns page bodies."
    )
    args_schema: Type[BaseModel] = SearchInput

    def _run(self, query: str, min_sources: int = 2) -> str:
        r = httpx.get(
            "https://api.unlob.com/search",
            headers={"x-api-key": os.environ["UNLOB_API_KEY"]},
            params={"q": query, "min_independent_sources": min_sources, "limit": 8},
            timeout=15,
        )
        r.raise_for_status()
        return "\n".join(
            f"[{h['independent_sources']} sources] {h['title']} — {h['url']}\n"
            f"  {h['snippet']}"
            for h in r.json()["results"]
        )


search = UnlobSearch()

researcher = Agent(
    role="Research analyst",
    goal="Find what is actually known about {topic}, with sources",
    backstory="You search before you assert, and you keep every URL.",
    tools=[search],
    llm="claude-opus-5",
)

checker = Agent(
    role="Fact checker",
    goal="Find claims that rest on a single source and flag them",
    backstory=(
        "You are sceptical by default. You re-search each claim with "
        "min_sources=3 and report what fails to corroborate."
    ),
    tools=[search],
    llm="claude-opus-5",
)

gather = Task(
    description="Research {topic}. Return findings, each with its URL.",
    expected_output="A bulleted list of findings, each ending with its source URL.",
    agent=researcher,
)

verify = Task(
    description=(
        "Take the findings and re-search each with min_sources=3. Mark any "
        "that fail to corroborate as UNVERIFIED and explain why."
    ),
    expected_output="The findings, each marked VERIFIED or UNVERIFIED with a reason.",
    agent=checker,
    context=[gather],   # the checker sees the researcher's output
)

crew = Crew(
    agents=[researcher, checker],
    tasks=[gather, verify],
    process=Process.sequential,
)

if __name__ == "__main__":
    result = crew.kickoff(
        inputs={"topic": "changes to EU AI Act GPAI rules in 2026"}
    )
    print(result)

What will go wrong

The failures that cost an afternoon rather than a minute, because they produce something that looks like it is working.

Where this agent can go next

Every result the agent receives is a node in the coverage graph. These are the tools you can add to it without leaving CrewAI — each is one more entry in the same tools list.

relatedNear-duplicatescorroborateIndependent sourcesauthoritiesAuthoritative hostsdossierThe entity briefpathA second documentassemble_contextA packed context setA search resultwhat your agent has
The two marked in blue are the tools this tutorial wires up. The other four are the same shape.

Frequently asked questions

Sequential or hierarchical process?

Sequential for a pipeline like this one, where each task feeds the next. Hierarchical adds a manager agent that delegates, which is worth it only when the task order genuinely depends on intermediate results.

Why return a formatted string from the tool rather than JSON?

CrewAI passes tool output into the prompt as text. A compact human-readable format costs fewer tokens than pretty-printed JSON and is easier for the model to read — which is why the source count leads each line here.

You need a key to run this

10,000 requests a month on the free tier, no card. Enough to build the agent and put a real evaluation set through it.