Staso Python SDK

Trace explicitly instrumented agent runs and evaluate proposed tool calls before your code executes them.

Supported launch configuration

• Python 3.11+
• OpenAI Chat Completions or Anthropic Messages
• Synchronous or asynchronous, non-streaming calls
• An explicit @st.agent root and @st.tool on executed Python tools
• Claude Code and Codex through one-command CLI setup
• Observe plus configured Guard policies in Audit or Enforce mode

Copy-paste launch sample

Use the canonical launch quickstart at https://staso.ai/docs/quickstart/launch. If you have a repository checkout, you can run the pinned copy in examples/launch_quickstart/:

pip install -r examples/launch_quickstart/requirements.txt

requirements.txt pins:

staso[openai]==0.5.15
openai==3.6.0

Set STASO_API_KEY, OPENAI_API_KEY, STASO_AGENT_NAME, STASO_ENVIRONMENT, STASO_WORKSPACE_SLUG, and STASO_SESSION_ID=launch-quickstart-first-trace, then run:

python examples/launch_quickstart/openai_guard_quickstart.py

The sample checks SDK delivery. Onboarding should wait for the dashboard/backend trace query before marking the first trace visible.

OpenAI quickstart

pip install "staso[openai]"

import json

import staso as st
from openai import OpenAI

st.init(api_key="ak_...", agent_name="weather-agent")
client = OpenAI()

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get the weather for a city.",
            "parameters": {
                "type": "object",
                "properties": {"city": {"type": "string"}},
                "required": ["city"],
            },
        },
    }
]


@st.tool
def get_weather(city: str) -> dict:
    return {"city": city, "temperature_c": 18}


@st.agent
def answer(question: str) -> dict | str:
    try:
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": question}],
            tools=TOOLS,
        )
    except st.GuardBlocked as blocked:
        return f"Blocked: {blocked.reason}"

    tool_call = response.choices[0].message.tool_calls[0]
    arguments = json.loads(tool_call.function.arguments)
    return get_weather(**arguments)


print(answer("What is the weather in London?"))
delivery = st.shutdown(timeout=5)
if not delivery.ok:
    raise RuntimeError(f"Trace delivery incomplete: {delivery.last_failure}")

st.init() patches an installed OpenAI SDK. The explicit decorators define the supported trace shape: @st.agent creates the root and @st.tool records the real tool duration, status, input, output, and error.

Guard evaluates tool proposals returned by the patched provider before the normal dispatch loop receives them. An audit finding records a would-block and lets the proposal continue. A block raises st.GuardBlocked.

@st.tool observes a Python function. It does not run Guard by itself. Check a direct tool call explicitly when it needs enforcement:

decision = st.guard("delete_record", {"record_id": "r_42"})
if decision.action == "block":
    raise RuntimeError(decision.reason)
delete_record("r_42")

Anthropic

Install staso[anthropic] and use a non-streaming messages.create call under the same explicit decorators. See the Anthropic quickstart.

CLI agents

Connect Claude Code or Codex without changing application code:

pip install staso
staso setup --target claude-code  # or codex

See the Claude Code quickstart, Codex quickstart, or CLI reference.

Pricing

Team starts at $39 USD per month for one unit. Each unit includes 10,000 monthly traces and 5,000 monthly Guard LLM-judge evaluations. Team includes unlimited static Guard evaluations.

Choose 1 to 50 Team units. See current pricing and limits for volume discounts and the final signed-in quote, currency, and included capacity. Spans do not add plan usage, but monthly allowances, request limits, and service rate limits still apply.

Capture boundaries

• requested_model records the request. model records the provider-resolved ID when the response supplies one, and otherwise falls back to the request. provider records the provider identity.
• usage_available=True means the provider returned usage, including valid zero-token values. False means the compatibility token counters are not evidence of known zero.
• Cost is estimated by the Staso service for recognized models. Unknown usage or an unrecognized model stays unknown instead of becoming $0.
• Five independent categories default on: request, system_prompt, tool_schema, tool_arguments, and output. Configure them with st.init(capture_<category>=...) or STASO_CAPTURE_<CATEGORY>.
• Suppression removes governed values before SDK payload serialization. Enabled content can still be redacted by the backend before storage; these are different states.
• The deprecated capture_messages=False fallback suppresses request and system_prompt only. Explicit category controls win.
• Span delivery is asynchronous. Call st.flush() to wait for events accepted before that call, or st.shutdown() when the process is done. Both return a DeliveryReport; check report.ok when delivery is required. report.rejected and report.last_rejection expose local queue drops that happened during automatic decorator cleanup.
• Guard fails open when no policy decision is available by default. Use fail_closed=True or STASO_GUARD_FAIL_CLOSED=true for calls that must block on explicit degradation.

For a warm serverless worker, flush after each invocation without stopping the shared client:

def handler(event, context):
    result = answer(event["question"])
    report = st.flush(timeout=2)
    if not report.ok:
        logger.warning("Staso delivery incomplete: %s", report.last_failure)
    return result

Calling st.init() again retires the previous client. Inspect new_client.previous_shutdown_report to verify that drain without making initialization throw by default.

Documentation

• Python decorators
• OpenAI Chat Completions
• Anthropic Messages
• Claude Code
• Codex
• CLI setup
• Guard audit and block
• SDK setup
• Troubleshooting

Python 3.11+ | Apache 2.0