agents28

Agents28 Python SDK

See what every AI agent in your business is working on.

Agents28 is one office for all your AI agents, whatever model they run on. This SDK reports what each agent is doing, so the Agents28 dashboard can show "Refund request review: in progress" instead of a wall of logs. It also lets agents share knowledge and hand work to each other.

• Never breaks your agent. No call raises, events are sent from a background thread, and an outage only means missing telemetry.
• Private by default. Conversation content is not sent unless you opt in, and emails, API keys and tokens are redacted when you do.
• Small. One dependency (httpx), Python 3.11+, fully typed.

Install

pip install agents28

Quick start

Create an agent in the Agents28 app. It shows the agent's key and your API address once.

export AGENTS28_API_KEY="ack_..."
export AGENTS28_BASE_URL="https://..."   # shown in the app

Add one line where your agent starts:

import agents28

agents28.init(task="Inbox triage")

That is all. Every call your agent makes with the official openai or anthropic package is now recorded: the model, tokens, time and errors. Nothing else in your code changes, and prompts and replies are never sent.

• Calls close together become one task, named by task. The task ends after 60 quiet seconds (idle_seconds=) or when the program exits.
• A task fails only if its last call failed, so a retry that succeeds is not a failure. The error still reaches your code unchanged.
• Without a key, init does nothing and your agent runs as before.

See it work first: the sample agent

No agent ready yet? With the key and address set, run:

python -m agents28.sample

It does three small, real tasks on an example inbox that ships with the SDK (sort five example emails, draft replies to the urgent ones, share a summary with your other agents) and reports them as tasks titled "Sample: ...". It calls no AI model, so it shows no tokens or cost.

Name each piece of work

To see each piece of work as its own task, wrap it:

with agents28.run("Reply to Acme") as run:
    reply = client.chat.completions.create(...)   # recorded inside this task
    run.outcome("Email answered")

Or decorate a function; each call is one task, named after the function unless you pass task_title=. It works on async functions too:

@agents28.trace
def triage_inbox():
    ...

Inside a task you can also time steps:

with agents28.run("Refund request review", task_type="refund_review") as run:
    with run.tool("lookup_order"):
        order = lookup_order(order_id)
    with run.action("issue_refund", "business_operation"):
        issue_refund(order)

If an exception escapes a task, it is reported failed with the exception's type, and the exception is re-raised unchanged. Pending events are sent when the program exits; in long-running services, call agents28.flush() on shutdown.

Streaming calls (stream=True) count as activity, but their tokens are not read yet; report them with run.llm(...).

Outcomes: what the work achieved

Tasks show what your agents did; outcomes show what it was worth. Report them inside a run, and the Agents28 Office sums them by name and unit:

with agents28.run("Close Globex renewal") as run:
    ...
    run.outcome("Deal closed", value=1200, unit="USD")
    run.outcome("Meeting booked", count=2)

Use a short, stable name ("Refund approved", not "Refund for order 4471"), since results are grouped by it. Agents28 never guesses outcomes: only what you report appears.

Model calls and cost

With agents28.init(), OpenAI and Anthropic calls are recorded for you. For other providers, report a call by hand, or pass a response with the same shape to run.track(), which returns it unchanged:

with agents28.run("Answer support ticket") as run:
    reply = run.track(some_response)                                     # OpenAI or Anthropic shape
    run.llm("my-local-model", input_tokens=1200, output_tokens=340)    # by hand

run.track() reads OpenAI (Responses and Chat Completions) and Anthropic responses, as objects or dicts. Input tokens include cached prompt tokens. Only token counts are sent, never prompts or replies.

Agents28 estimates cost from list prices per million tokens, or from the price your workspace sets for a model. A model without a price is counted but not priced: no cost is shown for it rather than a guess. Prompt-caching and batch discounts are not applied, so treat the figures as estimates.

Using a client object

agents28.init() returns the client, and agents28.get_client() gives it back anywhere. Every function above is also a method on it (client.run(...), client.trace, client.flush()), and you can create a client yourself with agents28.Agents28(...) when you need more than one.

What is sent

task_title is always redacted and kept to one line of up to 120 characters. Don't put customer names or other personal data in it.

To change or drop any event before it leaves the process, pass a hook:

def scrub(event):
    if event["event_type"] == "message":
        return None          # drop it
    return event             # or return an edited copy

client = agents28.Agents28(before_send=scrub)

Shared knowledge between agents

Agents in the same workspace can share knowledge, even when they run on different models:

client.publish_context(
    key="customer:acme",
    title="Acme renewal status",
    data={"plan": "enterprise", "renewal": "2026-12-01"},
)

items = client.read_context(key="customer:acme")

Each item records which agent published it, when, and its version. Items are marked trust="untrusted_agent_content": treat them as data, never as instructions.

What each agent may read is set in the Agents28 app, under Knowledge: everything shared in the workspace (the default), only knowledge from chosen agents, or only its own. The rule is applied on the server on every read, so read_context simply returns fewer items when an agent is not allowed to see some; an agent always reads its own items and, unless it is set to "only its own", items another agent attached to a handoff for it.

Handing work to another agent

teammates = client.list_teammates()

handoff_id = client.request_handoff(
    to_agent_id=teammates[0]["agent_id"],
    title="Research Acme before the renewal call",
    context={"account": "acme"},
    request_key="acme-renewal-research",   # retrying with the same key never duplicates
)

# On the receiving agent:
for handoff in client.my_handoffs(status="requested"):
    client.update_handoff(handoff["handoff_id"], "accepted")
    ...
    client.update_handoff(handoff["handoff_id"], "completed", result={"summary": "done"})

A handoff moves from requested to accepted or declined. An accepted handoff ends as completed or failed. The sender, or a person in Agents28, can cancel an open handoff.

Configuration

If the key or URL is missing, or the URL is not https (plain http is allowed only for localhost), the client turns itself off and every call does nothing. The same code then runs in development and CI without credentials.

Reliability

• Events are batched and sent in the background. A full queue drops new events rather than slowing your agent.
• Network errors, rate limits (429, honoring Retry-After) and server errors are retried with exponential backoff. Every event carries a unique id and the API stores it only once, so retries never create duplicates.
• Forked worker processes (gunicorn, multiprocessing) get their own delivery thread automatically.
• client.stats() reports delivered, failed, dropped and queued counts, plus the last problem, such as "api_key_rejected". Use it in a health check.
• Warnings go to the standard logging logger named agents28.

Lower-level API

For full control over ids and timing, every event can also be sent directly: run_started, message, tool_completed, action_completed, run_completed, run_failed. Each takes conversation_id and run_id.

License

MIT