langchain-blopus

This package contains the LangChain integration with Blopus, a web search and fetch API for LLMs and agents that runs on its own crawler and index rather than reselling another search engine.

pip install -U langchain-blopus

Set your API key, which you can create at blopus.ai/app:

export BLOPUS_API_KEY="blp_live_..."

Images

from langchain_blopus import BlopusSearch

tool = BlopusSearch(include_images=True)
out = tool.invoke({"query": "tesla factory"})
for r in out["results"]:
    if r.get("image"):        # absent is normal - coverage is partial
        print(r["title"], r["image"])

Image support is in BETA. Every result also carries word_count, so an agent can see that a hit is a stub before spending a fetch call on it.

Tools

BlopusSearch

Searches the live web and returns ranked results with snippets.

from langchain_blopus import BlopusSearch

tool = BlopusSearch()
tool.invoke({"query": "postgres logical replication", "freshness": "pw"})

{
    "query": "postgres logical replication",
    "results": [
        {
            "title": "...",
            "url": "https://...",
            "snippet": "...",
            "domain": "...",
            "score": 0.87,
            "published_at": 1755300000,
            "language": "en",
        }
    ],
}

Useful arguments: count (served in blocks of 10, so it rounds up), freshness (pd, pw, pm, p3m, p1y, all), include_domains, exclude_domains, language, news_only, and include_content to get the cleaned page text inline instead of making a second call.

BlopusFetch

Retrieves the full cleaned text of pages the model selected.

from langchain_blopus import BlopusFetch

BlopusFetch().invoke({"urls": ["https://example.com/article"]})

URLs that were not found come back in failed_urls rather than raising, so one bad URL in a batch does not lose the rest.

Using them with an agent

from langchain.agents import create_agent
from langchain_blopus import BlopusFetch, BlopusSearch

agent = create_agent(model, tools=[BlopusSearch(), BlopusFetch()])

Retriever

BlopusRetriever puts live web results into any chain written against the standard retriever interface, so it drops into an existing RAG pipeline in place of a vector store.

from langchain_blopus import BlopusRetriever

retriever = BlopusRetriever(k=10, freshness="pm")
docs = retriever.invoke("postgres logical replication")

docs[0].page_content        # cleaned article text
docs[0].metadata["source"]  # the URL

It requests full page content by default, because a Document whose page_content is a two-line snippet is close to useless for retrieval.

Async

Every tool and the retriever support ainvoke.

await BlopusSearch().ainvoke({"query": "postgres logical replication"})
await BlopusRetriever().ainvoke("postgres logical replication")

Notes

One topic per search. A query joining several subjects ("economy, sports, Iran") matches nothing, because no single document is about all of them. Issue one search per topic. The tool description states this, and the API returns a note when it detects the mistake, which this package passes straight through.

Billing is in blocks of ten. One search returns up to 10 results for 1 credit, so count rounds up to the next multiple of 10. If your remaining quota cannot cover the request, the API serves a partial page rather than refusing, and sets quota_clamped on the response so you can tell "my quota cut this short" apart from "that is all that exists".

License

MIT