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
Install
pip install crewai crewai-tools httpxAlready 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
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"] )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", )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
The whole thing
Complete and runnable. Set the two environment variables and it works.
"""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.
Pass the first task in the second task's `context`, or the fact-checker never sees the findings and cheerfully verifies nothing.
`expected_output` is not decoration — CrewAI uses it to decide when a task is done. Vague wording produces agents that ramble past the point.
Two agents is two sets of model calls. The fact-checker earns its cost here because it re-queries at a higher corroboration floor; a second agent that only rephrases the first is pure overhead.
Role-play prompting does not add rigour by itself. The scepticism in this crew comes from `min_sources=3` hitting the API, not from the word "sceptical" in the backstory.
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.
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.