backend_client

zocrates.mcp.backend_client

Backend client for Zocrates MCP Server.

Dispatches tool calls and resource reads over HTTP to the running Zocrates FastAPI backend service (:8765).

STRICT INVARIANT: Stdout is reserved strictly for JSON-RPC 2.0 protocol frames. All logging, diagnostic messages, and tracebacks MUST route exclusively to sys.stderr.

_handler module-attribute

_handler = _StderrStreamHandler()

logger module-attribute

logger = logging.getLogger('zocrates.mcp.client')

BackendClient

BackendClient(base_url=None, client=None, auth_token=None)

HTTP client for communicating with the local Zocrates FastAPI backend.

Initialize the MCP backend client.

Parameters:

Name	Type	Description	Default
base_url	str | None	Optional backend base URL (defaults to ZOCRATES_BACKEND_URL or http://127.0.0.1:8765).	None
client	Client | None	Optional preconfigured httpx.Client instance.	None
auth_token	str | None	Optional authentication token (auto-resolved if omitted).	None

Source code in zocrates/mcp/backend_client.py

def __init__(
    self,
    base_url: str | None = None,
    client: httpx.Client | None = None,
    auth_token: str | None = None,
) -> None:
    """Initialize the MCP backend client.

    Args:
        base_url: Optional backend base URL (defaults to ZOCRATES_BACKEND_URL or http://127.0.0.1:8765).
        client: Optional preconfigured httpx.Client instance.
        auth_token: Optional authentication token (auto-resolved if omitted).
    """
    self.base_url = (
        base_url or os.environ.get("ZOCRATES_BACKEND_URL", "http://127.0.0.1:8765")
    ).rstrip("/")
    self._auth_token = (
        auth_token.strip()
        if auth_token is not None
        else self._resolve_auth_token()
    )
    timeout = httpx.Timeout(connect=5.0, read=120.0, write=10.0, pool=5.0)
    self._client = client or httpx.Client(timeout=timeout)

close

close()

Close the underlying HTTP client.

Source code in zocrates/mcp/backend_client.py

def close(self) -> None:
    """Close the underlying HTTP client."""
    self._client.close()

get_catalog

get_catalog()

Retrieve root-level index.md catalog listing all converted library items.

Source code in zocrates/mcp/backend_client.py

def get_catalog(self) -> dict[str, Any]:
    """Retrieve root-level index.md catalog listing all converted library items."""
    return self._request("GET", "/mcp/catalog")

get_relations

get_relations(item_key, ceiling=25)

Retrieve verified related papers from the literature knowledge graph.

Source code in zocrates/mcp/backend_client.py

def get_relations(self, item_key: str, ceiling: int = 25) -> dict[str, Any]:
    """Retrieve verified related papers from the literature knowledge graph."""
    return self._request(
        "GET",
        f"/mcp/documents/{item_key}/relations",
        params={"ceiling": ceiling},
    )

get_toc

get_toc(item_key)

Retrieve Table of Contents and outline metadata for a document.

Source code in zocrates/mcp/backend_client.py

def get_toc(self, item_key: str) -> dict[str, Any]:
    """Retrieve Table of Contents and outline metadata for a document."""
    return self._request("GET", f"/mcp/documents/{item_key}/toc")

read_document

read_document(item_key, asset_id=None, page=None)

Read full document Markdown or targeted assets (sections, figures, tables).

Source code in zocrates/mcp/backend_client.py

def read_document(
    self,
    item_key: str,
    asset_id: str | list[str] | None = None,
    page: int | None = None,
) -> dict[str, Any]:
    """Read full document Markdown or targeted assets (sections, figures, tables)."""
    payload: dict[str, Any] = {}
    if asset_id is not None:
        payload["asset_id"] = asset_id
    if page is not None:
        payload["page"] = page

    return self._request("POST", f"/mcp/documents/{item_key}/read", json=payload)

search

search(query, item_key=None, chunk_type=None, limit=10, exclude_chunk_ids=None)

Perform hybrid semantic + BM25 keyword search over library chunks.

Source code in zocrates/mcp/backend_client.py

def search(
    self,
    query: str,
    item_key: str | list[str] | None = None,
    chunk_type: str | list[str] | None = None,
    limit: int = 10,
    exclude_chunk_ids: list[str] | None = None,
) -> dict[str, Any]:
    """Perform hybrid semantic + BM25 keyword search over library chunks."""
    payload: dict[str, Any] = {
        "query": query,
        "limit": limit,
    }
    if item_key is not None:
        payload["item_key"] = item_key
    if chunk_type is not None:
        payload["chunk_type"] = chunk_type
    if exclude_chunk_ids is not None:
        payload["exclude_chunk_ids"] = exclude_chunk_ids

    return self._request("POST", "/mcp/search", json=payload)

_StderrStreamHandler

Bases: StreamHandler

StreamHandler that dynamically resolves sys.stderr on emit.

stream property writable

stream

Dynamically resolve and return current sys.stderr stream.

Returns:

Type	Description
Any	The current sys.stderr stream object.