fathom-sdk 0.1.0

Python SDK for Fathom — typed client for the agent gateway over HTTP + SSE.

fathom-sdk

Python SDK for Fathom — a typed client for the agent gateway over HTTP + SSE. Sync and async.

Install

pip install fathom-sdk
# or uv pip install / poetry add / pdm add

Quick start (sync)

from fathom import Fathom

with Fathom(base_url="http://localhost:8790", token="...") as f:
    thread = f.threads.create()
    reply = f.threads.send(thread["id"], "What's the weather in NYC?")
    print(reply["agent_message"]["content"])

base_url and token can also come from FATHOM_BASE_URL and FATHOM_TOKEN env vars. If both are set you can call Fathom() with no args.

Async

import asyncio
from fathom import AsyncFathom

async def main() -> None:
    async with AsyncFathom() as f:  # picks up FATHOM_* env vars
        thread = await f.threads.create()
        async for evt in f.threads.stream(thread["id"]):
            if evt["type"] == "agent_done":
                print(evt["message"]["content"])
                break

asyncio.run(main())

Sync streaming works the same with a regular for loop:

for evt in f.threads.stream(thread["id"]):
    if evt["type"] == "agent_thinking":
        print("⏳ agent is working…")
    elif evt["type"] == "agent_done":
        print("✅", evt["message"]["content"])
        break

Pairing (for headless / CLI use)

If you don't have a token — say a one-shot CLI pairing against a relay-hosted gateway — call pair() once:

with Fathom(base_url="https://relay.darklake.ai/c/<rid>") as f:
    claim = f.pair(code="123456", device_name="ci-bot")
    authed = f.with_token(claim["token"])
    threads = authed.threads.list()

pair() is the only method that doesn't require a token; everything else raises PairingRequiredError.

Error handling

from fathom import Fathom, UnauthorizedError, HttpError

with Fathom() as f:
    try:
        f.threads.list()
    except UnauthorizedError:
        # token expired or wrong — re-pair / refresh
        ...
    except HttpError as err:
        print(err.status, err.body)

All errors derive from FathomError. Specific subclasses: UnauthorizedError, HttpError, PairingRequiredError, MalformedResponseError.

Reference

Method	Description
f.threads.list()	All non-deleted threads, newest first.
f.threads.get(id)	One thread + its message history.
f.threads.create(title="")	New thread; auto-titles on first message if blank.
f.threads.send(id, text)	Send a user message, return user + agent reply.
f.threads.delete(id)	Soft-delete (restorable).
f.threads.restore(id)	Undo a soft-delete.
f.threads.stream(id)	Iterator (sync) / AsyncIterator (async) of SSE events.
f.pair(code=..., device_name=...)	Claim a pairing code, returns token.
f.with_token(token)	Returns a new client with a different token, sharing connection pool.

Requirements

• Python 3.10+
• Depends on httpx — single dependency.

License

MIT — see LICENSE.