samcoadk

A small Agent Development Kit. Four ideas, and nothing underneath them.

Agent	one turn: build the prompt, call the model, run the tools it asks for, return the reply
Runner	one turn for a user in a session: load the history, call the agent, commit the result
SessionService	where conversations live — in memory, or in Mongo or MySQL
A2A	one agent calling another, in-process or over HTTP

The core is stdlib only. import samcoadk pulls no third-party module, and there is a test that says so.

Quickstart

from samcoadk import Agent, Runner

def order_status(order_id: str) -> str:
    """Look up the status of a customer order."""
    return f"Order {order_id} shipped on Tuesday."

agent = Agent(
    name="support",
    model="gateway/openai/gpt-4.1",      # or "echo" to run with no API key
    instruction="You are a support agent. Answer in one or two sentences.",
    tools=[order_status],
)

runner = Runner(agent)
print(runner.run(user_id="u1", session_id="s1", message="where is order A9?"))

Durable conversations are one line:

from samcoadk import DatabaseSessionService

runner = Runner(agent, session_service=DatabaseSessionService("mongodb://localhost:27017"))
runner = Runner(agent, session_service=DatabaseSessionService("mysql://user:pw@host/db"))

Behind HTTP:

pip install "samcoadk[server]"
samcoadk serve examples.support          # /run, /run_sse, /sessions, /a2a, agent card
samcoadk chat  examples.support          # the same Runner, in your terminal

Install

pip install samcoadk                 # the agent, in memory
pip install "samcoadk[mongo]"        # + Mongo sessions
pip install "samcoadk[mysql]"        # + MySQL sessions
pip install "samcoadk[server]"       # + the HTTP surface
pip install "samcoadk[all]"

MCP needs nothing installed — samcoadk.mcp speaks stdio and streamable HTTP over stdlib urllib.

The one rule

An Agent holds no conversation. Build one and share it across every request. The transcript lives in a Session, and Runner is what binds the two — so two callers cannot leak into each other's history, because there is no shared history to leak through.

runner.run(user_id="alice", session_id="s", message="my card is 4111…")
runner.run(user_id="bob",   session_id="s", message="what did I say?")   # sees nothing of alice

Sessions

A session holds events (what happened, in order) and state (what is true now — a phase, a draft, a counter; whatever your agent needs).

They are written together, once per turn:

runner.run(user_id="u1", session_id="s1", message="...", state_delta={"phase": "PLAN"})

One atomic checkpoint per turn means a crash cannot leave a transcript that says a question was asked next to a state that says it was not.

Another database

Implement five methods — ensure_schema, read, write, list, delete — and pass it in. All the session rules stay in DatabaseSessionService:

runner = Runner(agent, session_service=DatabaseSessionService(backend=MyRedisBackend()))

Guardrails

samcoadk ships none, deliberately: what counts as an unacceptable request is your agent's policy, not the kit's. The seam is callbacks.py.

from samcoadk import CallbackDecision

class NoSelfApproval:
    def after_agent(self, ctx, reply):
        if '"approved"' in reply:
            return CallbackDecision.block("approval is the caller's to give",
                                          replacement_result="Rejected: cannot self-approve.")

agent = Agent(name="planner", model=..., callbacks=[NoSelfApproval()])

before_agent is the input rail, after_agent the output rail, and before_tool / after_tool police individual tool calls. A hook may pass a value through, replace it, or block. Callbacks fail closed by default — set fail_open = True on one that is pure observability.

Agents calling agents

A remote agent and a local one satisfy the same interface — .name and .run(text) -> str:

from samcoadk import RemoteAgent

billing = RemoteAgent.discover("http://billing:8000")
reply = billing.run("what is invoice 42?")

Moving an agent out to its own service is invisible to its caller.

A handler that cannot continue raises Clarify, which the server reports as the A2A input-required state rather than a failure — the difference between "this agent is broken" and "this agent is waiting for you".

Two dialects, one method

message/send carries parts, and a part is either prose or a JSON object:

client.send("what is invoice 42?")                  # TextPart — a conversation
client.send(data={"op": "screen", "index": "NIFTY"})  # DataPart — a contract

Text is what you want when the answer is for a person and the caller cannot know the question in advance. Data is what you want between two programs that have agreed on fields: it needs no model to interpret it, so a peer can answer deterministically and prove it did.

from samcoadk.a2a import data_of, status_data, text_of

task = client.send(data={"op": "screen", "index": "NIFTY 50"})
result = data_of(task)          # the completed task's artifact DataPart

data_of is strict where text_of is forgiving, and deliberately: a thin text reply is a thin answer, but a caller that cannot tell "the screener is down" from "the screen matched nothing" will present an unresolved universe as a resolved one. So every way it can be wrong raises, and the exception says which. A paused task carries its payload on the status message instead — read it with status_data.

On the server, a DataPart goes through its own door:

def handle(payload: dict) -> dict:
    return {"matched": screen(payload["index"])}

app = create_app(runner, data_handler=handle)

The model still answers text. It never sees the DataPart — feeding a wire contract to a model asks it to re-derive what the payload already states, and makes a deterministic call cost a token budget and a round of nondeterminism. Without a data_handler configured, a DataPart is refused rather than silently stringified into a prompt.

Tests

pip install -e ".[test]"
pytest -q

Database tests skip silently without a server, so a bare green run does not mean they executed:

SAMCOADK_TEST_MONGO=mongodb://localhost:27017 \
SAMCOADK_TEST_MYSQL='mysql://user:pw@localhost/samcoadk_test' pytest -q

Coming from 0.2.0

0.2.0	0.3.0
agent.run(text, history=session.history)	runner.run(user_id=…, session_id=…, message=…)
Agent(...).history	gone — the transcript is Session.events
ConversationHistory	Session.events + session.as_messages()
samcoadk.sessions.InMemorySessionStore	InMemorySessionService
samcoadk.mongo.MongoSessionStore	DatabaseSessionService("mongodb://…")
samcoadk.Guardrails, keyword_blocklist, …	your own callbacks (see above)
samcoadk.runtime.* (capabilities, contracts, boot)	gone — build the agent in Python
boot.load(__file__, policy=…, models=…)	Agent(...) and Runner(...)
samcoadk.a2a.registry.AgentRegistry	RemoteAgent.discover(url)
samcoadk.console.create_app	samcoadk.server.create_app(runner)
Agent(model="openai/gpt-4.1") + gateway env	Agent(model="gateway/openai/gpt-4.1") — explicit
samcoadk.ops.* (progress, prometheus, budget)	gone — Metrics/Tracer remain in observability

The gateway is no longer selected by an environment variable being set. Say gateway/ and pass the URL, or the same model string means different things in different shells.