agent-web-search 0.1.0

Python client for the self-hosted Agent Web Search API — returns cosine-ranked, scrubbed, RAG-ready chunks

agent-web-search

Python client for the Agent Web Search API — self-hosted, RAG-ready web search with cosine-ranked chunks.

Returns scrubbed, chunked, cosine-ranked text from a 6-stage pipeline (SearXNG → fetch → extract → scrub → chunk → embed → rank). Drop-in for Tavily / Exa in self-hosted setups — same shape, no API keys, no usage limits, your data stays on your machine.

Why this SDK?

	agent-web-search	Tavily	Exa
Cost	Free, self-hosted	$0.05/query	$0.005/query
API key	None	Required	Required
RAG chunk output	✅ 256/512-token with scores	❌	❌
Prompt scrubbing	✅ Built-in	❌	❌
JS rendering	✅ Opt-in	❌	❌
Type-safe	✅ Pydantic v2	Partial	Partial
Sync + async	✅	✅	✅

The agent-web-search server is the full pipeline; this client is a thin, type-safe wrapper with sync and async surfaces, Pydantic models for every response, and an exception hierarchy that maps cleanly to OpenAI's SDK pattern.

Install

pip install agent-web-search

Zero required dependencies beyond httpx and pydantic (both installed automatically).

Quick start

Sync

from agent_web_search import AgentWebSearch

with AgentWebSearch(base_url="http://localhost:8000") as client:
    response = client.search("python asyncio", include_content=True)
    for chunk in response.ranked_chunks:
        print(f"{chunk.score:.3f}  {chunk.source.title}")
        print(f"        {chunk.source.url}")
        print(f"        {chunk.text[:120]}...")

Async

import asyncio
from agent_web_search import AsyncAgentWebSearch

async def main():
    async with AsyncAgentWebSearch(base_url="http://localhost:8000") as client:
        response = await client.search("python asyncio")
        print(response.top_chunk.text if response.top_chunk else "no results")

asyncio.run(main())

Concurrent queries

import asyncio
from agent_web_search import AsyncAgentWebSearch

async def main():
    async with AsyncAgentWebSearch() as client:
        results = await asyncio.gather(
            client.search("python asyncio"),
            client.search("rust borrow checker"),
            client.search("docker healthcheck"),
        )

asyncio.run(main())

API reference

AgentWebSearch(base_url=None, *, timeout=30.0, headers=None) (and async)

Create a client. All arguments are optional — sensible defaults work out of the box if the server is at http://localhost:8000.

Argument	Type	Default	Description
base_url	str	AGENT_WEB_SEARCH_BASE_URL env var, else http://localhost:8000	Server root URL
timeout	float	30.0 seconds	Per-request timeout (covers the full pipeline)
headers	dict[str, str]	None	Extra HTTP headers (e.g. future auth tokens)

Both clients are context managers (sync: with, async: async with) and can be used without one too (just call close() / await close() when done).

client.search(query, *, render_js=False, include_content=False, categories=None, time_range=None, timeout=None)

Search the web and return cosine-ranked chunks.

Argument	Type	Default	Description
query	str	required	1–1000 characters
render_js	bool	False	Use crawl4ai to render JS-heavy pages (slower; opt in only when needed)
include_content	bool	False	Fetch, extract, chunk, and rank each page. If False, ranks pre-fetched SearXNG snippets
categories	list[str]	None	Override intent routing — e.g. ["it"], ["news"], ["general"]
time_range	str	None	Time filter for news queries: "day", "week", "month", "year"
timeout	float	client default	Per-call timeout override

Returns: SearchResponse with query, ranked_chunks: list[RankedChunk], unresponsive_engines: list[str].

Raises: APITimeoutError, APIConnectionError, APIError, APIResponseError, or ValueError for invalid input.

client.health(*, timeout=None)

Ping the server. Returns HealthResponse(status="ok") on success.

Response models

SearchResponse(
    query="python asyncio",
    ranked_chunks=[
        RankedChunk(
            text="asyncio is a library...",            # ~256-token chunk
            parent_text="asyncio — Asynchronous I/O...", # ~512-token context
            chunk_index=0,
            score=0.87,                                # cosine similarity
            source=ChunkSource(
                url="https://docs.python.org/3/library/asyncio.html",
                title="asyncio — Asynchronous I/O — Python 3.12 documentation",
                searxng_score=1.0,                     # upstream popularity
            ),
        ),
        ...
    ],
    unresponsive_engines=["duckduckgo"],                # engines that timed out
)

SearchResponse has two convenience properties:

• response.top_chunk — the highest-scoring chunk, or None if empty
• response.urls — the unique source URLs, in ranking order

Configuration

Env var	Default	Description
AGENT_WEB_SEARCH_BASE_URL	http://localhost:8000	Server URL (overridden by base_url kwarg)
AGENT_WEB_SEARCH_TIMEOUT	30.0	Default request timeout in seconds

Errors

All exceptions inherit from AgentWebSearchError. The hierarchy mirrors the OpenAI SDK so error-handling patterns are portable:

Exception	When it's raised	Has
APITimeoutError	Request timed out (client-side timeout exceeded)	—
APIConnectionError	Couldn't reach the server (DNS, refused, network)	—
APIError	Server returned non-2xx (4xx, 5xx)	status_code, body, detail
APIResponseError	Server returned a body that didn't match the expected schema	—

from agent_web_search import (
    AgentWebSearch, AgentWebSearchError,
    APIError, APITimeoutError, APIConnectionError,
)

client = AgentWebSearch(base_url="http://localhost:8000")

try:
    response = client.search("python asyncio", timeout=15.0)
except APITimeoutError:
    print("The server is too slow — try increasing timeout")
except APIConnectionError:
    print("Server is not running. Start it with: docker compose up")
except APIError as e:
    print(f"Server returned HTTP {e.status_code}: {e.detail}")
except AgentWebSearchError as e:
    # Catch-all for any other SDK error
    print(f"SDK error: {e}")

Development

git clone https://github.com/blueewhitee/agent-web-search-sdk
cd agent-web-search-sdk

# Install dev dependencies
uv sync --group dev

# Run tests
uv run pytest -v

# Type check
uv run mypy src  # (if mypy is installed)

# Build sdist + wheel
uv build

# Publish to PyPI (one-time, requires API token)
export UV_PUBLISH_TOKEN="pypi-AgEIcHlwaS5vcmcC..."  # token from pypi.org
uv publish

Publishing (release process)

The SDK uses PyPI Trusted Publishing via OIDC + GitHub Actions — no API tokens stored anywhere. Here's the one-time setup:

One-time setup

1. Create the GitHub repo (e.g. github.com/blueewhitee/agent-web-search).
2. Push the SDK code (this directory) to the repo.
3. Register a Trusted Publisher on PyPI:
   • Go to https://pypi.org/manage/project/agent-web-search/settings/publishing/ (or create the project first if it doesn't exist).
   • Add a pending publisher (or normal publisher after the first release):
     • Owner: blueewhitee
     • Repository: agent-web-search
     • Workflow filename: publish.yml
     • Environment name: pypi
4. Optional first release manually (recommended, to lock in the name immediately):

   # Create a PyPI API token at https://pypi.org/manage/account/token/
   # Scope it to the agent-web-search project only.
   export UV_PUBLISH_TOKEN="pypi-..."
   uv build
   uv publish
   

5. Switch the PyPI publisher from "pending" to "live" now that the project exists.

Every release after that

# Bump version in pyproject.toml and src/agent_web_search/__init__.py
git commit -am "release: v0.2.0"
git tag v0.2.0
git push --tags

# GitHub Actions automatically builds + publishes to PyPI via OIDC.

The workflow file at .github/workflows/publish.yml handles the rest.

Architecture

This is intentionally a thin client (~250 LOC of production code):

• src/agent_web_search/client.py — sync AgentWebSearch
• src/agent_web_search/async_client.py — async AsyncAgentWebSearch
• src/agent_web_search/models.py — Pydantic v2 models
• src/agent_web_search/exceptions.py — exception hierarchy
• src/agent_web_search/_base.py — shared config + httpx→SDK error translation

The models are re-declared in the SDK rather than imported from the server to keep the package self-contained and the dependency surface minimal. The FastAPI server generates an OpenAPI schema at /openapi.json; a v0.2 follow-up could add codegen (datamodel-code-generator) to keep the two in sync automatically.

License

MIT