Instrument the execution boundary

Use @st.agent on the agent entry point and @st.tool on Python functions that perform tool work. Both decorators support regular and async def functions.

Install and initialize

pip install staso
export STASO_API_KEY=ak_...

import staso as st

st.init(agent_name="support-agent")

agent_name and the API key are required. You can pass the key directly as api_key="ak_...".

Decorate the agent and tools

import staso as st


@st.tool
def fetch_account(account_id: str) -> dict:
    return db.accounts.get(account_id)


@st.agent
def handle_request(account_id: str) -> str:
    account = fetch_account(account_id)
    return f"Account status: {account['status']}"

@st.agent creates the supported root span. @st.tool surrounds the real function call, so its duration, status, input, output, and exception come from the actual execution.

Guard direct tool calls explicitly

@st.tool records execution. It does not evaluate Guard by itself. Check a direct call before running it when the action needs enforcement:

@st.tool
def delete_record(record_id: str) -> None:
    db.records.delete(record_id)


@st.agent
def remove(record_id: str) -> str:
    decision = st.guard("delete_record", {"record_id": record_id})
    if decision.action == "block":
        return f"Blocked: {decision.reason}"
    delete_record(record_id)
    return "Deleted"

Patched OpenAI and Anthropic integrations evaluate provider-proposed tool calls before the normal dispatch loop receives them. See the OpenAI and Anthropic quickstarts.

Provider patches observe proposals and matching continuations. They do not prove that your application executed the tool. Keep @st.tool on the function that performs the work.

Capture controls

Set capture_input=False or capture_output=False on a decorator to omit that function payload. This suppresses capture; it does not transform or redact the value.

@st.tool(capture_input=False)
def look_up_secret(secret_id: str) -> str:
    return vault.read(secret_id)

Confirm delivery

Short-lived processes should shut down after the final run and inspect the structured result:

report = st.shutdown(timeout=5)
if not report.ok:
    raise RuntimeError(f"Trace delivery incomplete: {report.last_failure}")

Use st.flush(timeout=5) instead in a warm serverless worker. It waits for events accepted before the call and keeps the client ready for the next invocation.

DeliveryReport.ok is false for delivery failures, pending events, timeouts, or local rejections. rejected counts spans the local queue could not accept, including spans ended automatically by decorators. last_rejection provides the safe failure category.