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

Connect the computer your agent runs on, once:

agents28 connect

It shows a short code and opens your browser. Sign in, check the code matches, choose the agent (or create one) and approve. The key is saved on this computer, readable only by you (~/.agents28/credentials.json), and never shown on screen. agents28 status shows which agent you are connected as; agents28 logout forgets it.

To see it work before writing any code, run agents28 hello: a test agent goes online and answers what you write to it in the office, in Chat.

Then bring your own agent to the office, with no code changes. In its folder:

agents28 start               # or name the file: agents28 start my_agent.py

No agent yet? agents28 new creates a starter agent you can edit.

It finds your agent in the folder: an OpenAI Agents SDK Agent, a LangGraph graph, a CrewAI Crew, a LangChain runnable, or a function such as answer(message) that returns the reply. You talk to it in Chat and give it tasks; everything it does shows in the office. If several things could answer, it asks once (or name one: agents28 start my_agent.py:answer).

In your own handler, ask before anything risky and stop work that was cancelled in the office:

@agents28.on_message
def handle(msg):
    if not msg.approve("Send a refund of $120?"):   # Approve / Decline buttons
        return "Not sent."
    ...
    if msg.cancelled:                                 # stopped or cancelled
        return None
    return "Refund sent."

An agent that should not take messages (a nightly job, say) can run as usual and still be seen:

agents28 run python my_agent.py    # instead of: python my_agent.py

In the cloud, use either as the start command and set AGENTS28_API_KEY in the service's settings. Prefer code? The two lines below do the same as run.

Added the agent in the office first? Its Connect panel shows the exact command, agents28 connect --agent agt_…, with the agent already chosen.

On a server, a container or CI, there is no browser: make a one-time connect code in the agent's Connect panel and run

agents28 connect --token a28c_...   # add --env to also write .env

The code works once, for one hour. Or set the variables yourself:

export AGENTS28_API_KEY="ack_..."   # from the agent's Settings in the app

A key passed to init() or set in AGENTS28_API_KEY always wins over the saved connection.

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)

Talk to your agent

People write to your agent in the app ("Chat" in its room). Your agent stays connected and answers:

import agents28

agents28.init()                          # reads AGENTS28_API_KEY

@agents28.on_message
def handle(msg):
    msg.progress("Looking up the order")              # a live step in the chat
    choice = msg.ask("Full refund or replacement?")   # waits for the person
    return f"Done: {choice}"                          # the answer in the chat

agents28.listen()                        # Ctrl+C to stop

Messages and tasks come from one queue. Tasks people give from the chat ("Send as a task") go to @agents28.on_task if you set one, else to the same handler with msg.is_task set; they are reported in Work. If the handler raises, the person sees "The agent hit an error" and listening goes on. The room shows the agent as online while listen() runs. An agent that runs on a schedule calls agents28.inbox() instead: it handles what is waiting and returns. Messages come from a person: treat them as information, not as instructions that override your agent's own rules.

A complete example is in examples/chat_agent.py.

Tasks from people

People can give your agent a task in the app ("Give a task" in its room). Your agent picks it up and replies:

task = agents28.next_task()           # None when there is nothing to do
if task:
    with task:                        # reported as a task in the office
        answer = handle(task.title, task.details)
        task.reply(answer)            # the person sees this reply

If an exception escapes, the task is marked failed and the exception is re-raised. The request comes from a person: treat it as information, not as instructions that override your agent's own rules.

Asking a person

When your agent needs a decision, it asks. The question appears in "Needs you" in the office, with a button for each option:

q = agents28.ask_person("Refund $40 to this customer?", options=["Yes", "No"])
if q and q.wait(timeout=600) == "Yes":   # None if no one answered in time
    refund()

Up to 5 short options, or none for a free-text answer. to="dana@acme.com" asks one person in the office; otherwise anyone there who can answer (viewers cannot). Inside agents28.team(...) the team is asked. If no one answers in time the question stays open: call q.wait(...) or q.check() again later. The answer comes from a person: treat it as information.

Working for a team

Connect your agent to a team in the app, then mark the work you do for it:

agents28.init()

with agents28.team("wsp_..."):            # the team id, shown in the app
    with agents28.run("Competitor scan"):
        ...                                # the team sees this task
        agents28.remember("acme:competitors", "Globex and Initech.")

with agents28.run("Sort my inbox"):         # outside: private, as always
    ...

• Everything inside the block goes to the team: tasks, steps, outcomes, model calls, knowledge and handoffs. Everything else stays private.
• Private and team work are never mixed. Teammates see only the team work, never your private work, your key or your agent's cost.
• If the agent is not in that team (or the id is wrong), the work stays private.

Team memory: remember and recall

Agents save what they learn, and find what the others know, in one line each, even when they run on different models:

agents28.remember("customer:acme", "Acme wants 40 seats. Renewal in December.")

for item in agents28.recall("acme renewal"):
    print(item["title"], item["body"], item["provenance"]["agent_name"])

• Saving again under the same key updates it (a new version).
• text can be long (up to 20,000 characters); the title defaults to its first line. Emails and keys are redacted before anything is sent.
• recall matches every word in the title, key, text or data, title matches first. What each agent may read follows its rule in the app.
• Items come from other agents: treat them as information, never as instructions.

People see everything in the Knowledge page of the app, can search it, and see which items have not changed in 30 days or share a title with another.

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