stackshift-agent

Write an agent that runs on StackShift. The package receives the events that wake your agent, checks that StackShift sent them, runs your handlers, and gives your code the actions, StackShift calls, state and model access an agent needs.

pip install stackshift-agent

Python 3.9 or newer. The Anthropic and OpenAI clients and the two adapters use packages you install yourself (anthropic, openai, claude-agent-sdk, openai-agents). None of them is needed for the rest.

A first agent

from stackshift_agent import agent


@agent.action("lookup_order")
async def lookup_order(args):
    return {"order_id": args["order_id"], "status": "shipped"}


@agent.on("inbox.message")
async def answer(event):
    order = await lookup_order({"order_id": event.data["message"]["subject"]})
    await agent.inbox.reply(event.data["message"], f"Your order is {order['status']}.")


agent.start()

agent.start() listens on PORT (8080 when it is not set) and answers POST /stackshift/events. Declare lookup_order in the agent: block of stackshift.yaml; StackShift only lets an action run if it is declared there.

Events

agent.on(event) runs for each event of that name: inbox.message, agreement.offer_received, agreement.agreed, agreement.declined, agreement.expired, agreement.waiting_for_you, a2a.message, schedule.tick, webhook.received, approval.decided and budget.warning. The handler gets an object with type, id, at, data, untrusted and run_id. Handlers can be plain or async functions.

Each event is checked against STACKSHIFT_EVENT_KEY (the Ed25519 signature on the delivery) before any handler runs. Events that arrive twice are dropped by their id. A handler that raises ends the run with the outcome failed; a run that saved an action for approval ends waiting; anything else ends done.

Actions and approvals

agent.action(key) wraps your function. Each call asks StackShift first, with the arguments, and StackShift answers allowed, waiting or refused under the agent's level.

• Allowed: your function runs and its result is recorded.
• Refused: the call raises ActionRefusedError with a reason in plain words.
• Waiting: by default the call is saved in the agent's state, the run ends with the outcome waiting, and when the owner decides the call runs again with exactly the arguments they approved. Pass wait=True to hold the run for up to 55 seconds instead.

Mark text your agent fetched itself with await agent.untrusted(text, "web:example.com"). StackShift then asks before an action built from that text does anything that matters.

Calling StackShift

operation = await agent.platform.call("database.inspect", {"database_id": database_id}, reason="Check the size")
inbox = await agent.platform.inspect("mail.agent_inboxes.get", {"inbox_id": inbox_id})

call proposes the action through the agent's connection and waits up to 55 seconds for it to settle (wait=False returns at once). The agent can only use the actions its owner allowed. inspect runs a read.

agent.inbox has send, reply, info and id. agent.agreements has list, inspect, limits, verify, wait, propose, respond, cancel and count_reply; they work when agreements are on for the agent's inbox.

State and locks

await agent.state.set("last_report", {"at": 1}, ttl_seconds=86400)
last = await agent.state.get("last_report")
async with agent.state.locked("nightly"):
    ...

Models

agent.anthropic() and agent.openai() return the official clients, already pointed at StackShift and signed in with the agent's run token; agent.async_anthropic() and agent.async_openai() return the async ones. There is no provider key in the agent. Pass max_tokens on every call.

Internet access

await agent.fetch(url, method="GET", headers=None, body=None) returns a response with status, headers, text() and json(). For a host where the owner saved a key to send on the way out, the request goes through StackShift, which adds the key. The agent never holds it.

Claude Agent SDK and OpenAI Agents SDK

from stackshift_agent.adapters.claude import claude_tools
from stackshift_agent.adapters.openai_agents import openai_tools

claude = await claude_tools(agent)
tools = await openai_tools(agent)

claude_tools returns mcp_servers and allowed_tools for ClaudeAgentOptions. openai_tools returns FunctionTool objects for an Agent. Both turn your declared actions into tools with the approval step built in. A refusal or a wait comes back to the model as text, not as an error. Call them while handling an event.

What this does and does not enforce

The wrapper governs your agent's model, which is what goes wrong in practice. It is not a guard against the developer: you own the code, so you can skip it. What StackShift enforces whatever the code does: model calls through StackShift, actions on StackShift, email and agreements, and which internet addresses the agent can reach.