unlob Docs
Console
Browse documentation

CrewAI

A research tool for a research crew.

A tool

CrewAI tools are classes with a Pydantic argument schema.

import os, httpx
from typing import Type
from crewai.tools import BaseTool
from pydantic import BaseModel, Field

class SearchInput(BaseModel):
    query: str = Field(description='Supports AND, OR, -exclude and "exact phrase".')
    vertical: str | None = Field(None, description="e.g. code, science. Omit to auto-route.")
    limit: int = Field(5, description="Maximum results, 1-20.")

class UnlobSearch(BaseTool):
    name: str = "unlob_search"
    description: str = (
        "Search the web for passages. Returns passage text, not links, so the result "
        "is usually the answer rather than something to fetch. Each hit carries "
        "independent_sources: how many distinct sites assert it."
    )
    args_schema: Type[BaseModel] = SearchInput

    def _run(self, query: str, vertical: str | None = None, limit: int = 5) -> str:
        params = {
            "q": query,
            "limit": min(limit, 20),
            # Applied here rather than left to the agent: drop single-source claims,
            # and fold near-duplicates so the crew does not read one article five times.
            "min_independent_sources": 2,
            "collapse": "story",
        }
        if vertical:
            params["vertical"] = vertical

        r = httpx.get(
            "https://api.unlob.com/search",
            params=params,
            headers={"x-api-key": os.environ["UNLOB_API_KEY"]},
            timeout=20,
        )
        if r.status_code == 429 and "retry-after" not in r.headers:
            return "Monthly quota exhausted. Do not retry; report this to the user."
        r.raise_for_status()
        body = r.json()

        header = ("WARNING: incomplete results — part of the corpus was unreachable.\n\n"
                  if body.get("partial") else "")
        hits = "\n\n".join(
            f"{h['title']} — {h['url']} ({h.get('independent_sources', 0)} independent sources)\n{h['snippet']}"
            for h in body["results"]
        )
        return header + (hits or "No results found.")

A verification tool worth pairing with it

The reason to use unlob in a crew rather than a generic search tool:

class CorroborateInput(BaseModel):
    id: str = Field(description="A passage id from a search result.")

class UnlobCorroborate(BaseTool):
    name: str = "unlob_corroborate"
    description: str = (
        "Check whether a claim is independently reported. Given a passage id, returns "
        "the distinct hosts carrying that story, grouped and ranked by authority. Use "
        "this before stating anything as established fact — forty results echoing one "
        "source and four independent reports look identical in a result list."
    )
    args_schema: Type[BaseModel] = CorroborateInput

    def _run(self, id: str) -> str:
        r = httpx.get(
            "https://api.unlob.com/corroborate",
            params={"id": id},
            headers={"x-api-key": os.environ["UNLOB_API_KEY"]},
            timeout=20,
        )
        r.raise_for_status()
        b = r.json()
        hosts = ", ".join(s["host"] for s in b["sources"])
        return (f"{b['independent_sources']} independent sources "
                f"({b['merged_duplicates']} duplicates folded in): {hosts}")

Using them

from crewai import Agent, Task, Crew

researcher = Agent(
    role="Research analyst",
    goal="Find what is actually established about {topic}, and what is only asserted",
    backstory=(
        "You distinguish corroborated fact from repetition. You never state something "
        "as established without checking how many independent sources carry it."
    ),
    tools=[UnlobSearch(), UnlobCorroborate()],
)

task = Task(
    description="Research {topic}. Corroborate every claim you intend to report.",
    expected_output="A brief listing each finding with its independent source count.",
    agent=researcher,
)

Crew(agents=[researcher], tasks=[task]).kickoff(inputs={"topic": "…"})

The backstory is doing real work there. A crew given a search tool will search; a crew told what distinguishes a fact from a repetition will use the second tool.

Next