unlob Docs
Console
Browse documentation

LangChain and LangGraph

The MCP adapter for all eleven tools, or one thin tool if you want control.

Through MCP

langchain-mcp-adapters turns an MCP server into LangChain tools, so you get all eleven without writing schemas.

pip install langchain-mcp-adapters langchain-anthropic langgraph
import os
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_anthropic import ChatAnthropic
from langgraph.prebuilt import create_react_agent

client = MultiServerMCPClient(
    {
        "unlob": {
            "transport": "streamable_http",
            "url": "https://api.unlob.com/mcp",
            "headers": {"x-api-key": os.environ["UNLOB_API_KEY"]},
        }
    }
)

tools = await client.get_tools()
agent = create_react_agent(ChatAnthropic(model="claude-opus-5"), tools)

result = await agent.ainvoke(
    {"messages": [("user", "What is independently corroborated about the Acme acquisition?")]}
)

The tool descriptions come from the server, so the model already knows that corroborate answers “do independent sources say this” and that assemble_context returns reading rather than links. You do not need to explain the tools in your prompt.

One tool, hand-written

Eleven tools is a lot of schema in a context window. If your agent only ever searches, a single tool is cheaper and gives the model less to get wrong.

import os, httpx
from langchain_core.tools import tool

_client = httpx.Client(
    base_url="https://api.unlob.com",
    headers={"x-api-key": os.environ["UNLOB_API_KEY"]},
    timeout=20,
)

@tool
def web_search(query: str, vertical: str | None = None, limit: int = 5) -> str:
    """Search the web for passages. Returns passage text, not links.

    Results are already deduplicated and trust-ranked. Each carries
    `independent_sources`: how many distinct sites assert it.
    """
    params = {"q": query, "limit": limit, "min_independent_sources": 2}
    if vertical:
        params["vertical"] = vertical
    r = _client.get("/search", params=params)
    r.raise_for_status()
    body = r.json()

    # Say it out loud. A short answer here means part of the corpus was
    # unreachable, and a model given no warning will report it as "nothing found".
    prefix = "WARNING: incomplete results.\n" if body.get("partial") else ""
    return prefix + "\n\n".join(
        f"{h['title']} — {h['url']} ({h.get('independent_sources', 0)} sources)\n{h['snippet']}"
        for h in body["results"]
    ) or "No results."

As a retriever

For a RAG chain that expects Documents:

from langchain_core.documents import Document
from langchain_core.retrievers import BaseRetriever

class UnlobRetriever(BaseRetriever):
    k: int = 8

    def _get_relevant_documents(self, query: str, **_) -> list[Document]:
        r = _client.get("/search", params={
            "q": query,
            "limit": self.k,
            # One row per near-duplicate cluster. Without this, a chunk of your
            # context is the same wire story five times.
            "collapse": "story",
        })
        r.raise_for_status()
        return [
            Document(
                page_content=h["snippet"],
                metadata={k: h.get(k) for k in
                          ("url", "host", "title", "independent_sources", "host_rank")},
            )
            for h in r.json()["results"]
        ]

No chunking, no embedding, no vector store: hits are already passages.

Skipping the retrieval chain

assemble_context does search, dedupe, corroborate, rank and pack server-side, so for “what should I know about X” you can replace the chain with one call:

r = _client.get("/assemble_context", params={"q": question, "budget": 4000})
pack = r.json()
context = "\n\n".join(f"[{i['reason']}] {i['hit']['snippet']}" for i in pack["items"])

One billed request instead of several, and the reason on each item is a citation hint worth passing through to the model.

Next