Anthropic Messages

This launch path supports synchronous and asynchronous, non-streaming messages.create calls under an explicit @st.agent root.

For the pinned default onboarding sample, start with Launch quickstart. To pin this Anthropic path:

staso[anthropic]==0.5.15
anthropic==1.3.0

Use messages.create, claude-haiku-4-5, explicit decorators, and the same configured Guard Audit/Enforce boundary.

Install

pip install "staso[anthropic]"
export STASO_API_KEY=ak_...

Synchronous example

import staso as st
from anthropic import Anthropic

st.init(agent_name="weather-agent")
client = Anthropic()

TOOLS = [
    {
        "name": "get_weather",
        "description": "Get the weather for a city.",
        "input_schema": {
            "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.messages.create(
            model="claude-haiku-4-5",
            max_tokens=256,
            messages=[{"role": "user", "content": question}],
            tools=TOOLS,
        )
    except st.GuardBlocked as blocked:
        return f"Blocked: {blocked.reason}"

    tool_use = next(block for block in response.content if block.type == "tool_use")
    return get_weather(**tool_use.input)

st.init() patches the installed Anthropic SDK. @st.agent supplies the supported root; @st.tool records the real tool call.

Asynchronous example

import staso as st
from anthropic import AsyncAnthropic

client = AsyncAnthropic()


@st.agent
async def answer_async(question: str) -> str:
    response = await client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=256,
        messages=[{"role": "user", "content": question}],
    )
    return "".join(block.text for block in response.content if block.type == "text")

Do not pass stream=True or use messages.stream(...) on the supported launch path.

Guard behavior

When a non-streaming response contains tool proposals, Staso evaluates configured Guard policies before the response reaches the normal dispatch loop.

• Audit findings emit a guard:would-block:<tool> child span and allow the response through.
• A block emits a guard:blocked:<tool> child span and raises st.GuardBlocked.
• @st.tool records execution after an allowed proposal is dispatched. It does not run Guard itself.

Guard fails open when no policy decision is available by default. See manual checks for the optional fail-closed setting and stable failure fields.

Captured fields

The LLM span records supported fields exposed by the provider SDK, plus token usage when the response includes it. Cost is estimated by the Staso service for recognized models and may be unavailable.

Tool names/IDs, model, token counts, latency, status, stop reason, and response ID can remain as operational correlation data when their content category is suppressed. Tool input embedded in assistant output requires both tool_arguments and output.

All categories default to enabled. Configure them with st.init(capture_<category>=...) or STASO_CAPTURE_<CATEGORY>. The deprecated capture_messages=False suppresses both user messages and the standalone system prompt. See SDK setup for full precedence.

Next

• OpenAI Chat Completions
• Python decorators
• Guard