Python SDK Tracing

Overview

The Anyway SDK provides two decorators to structure your traces:

@workflow - High-level operations that orchestrate multiple tasks
@task - Individual units of work (like LLM calls)

These work identically with any LLM provider.

Basic Setup

from anyway.sdk import Traceloop
from anyway.sdk.decorators import workflow, task

Traceloop.init(app_name="my-app")

The @workflow Decorator

Use @workflow for top-level operations that coordinate multiple steps:

@workflow(name="process_document")
def process_document(doc: str) -> dict:
    summary = summarize(doc)
    keywords = extract_keywords(doc)
    return {"summary": summary, "keywords": keywords}

Workflows create parent spans that contain all nested operations.

The @task Decorator

Use @task for individual operations, especially LLM calls:

@task(name="summarize")
def summarize(text: str) -> str:
    # Your LLM call here
    return result

Tasks create child spans within workflows.

Complete Example

OpenAI
Anthropic

from anyway.sdk import Traceloop
from anyway.sdk.decorators import workflow, task
from openai import OpenAI

Traceloop.init(app_name="document-processor")
client = OpenAI()

@task(name="summarize")
def summarize(text: str) -> str:
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": "Summarize the text concisely."},
            {"role": "user", "content": text}
        ]
    )
    return response.choices[0].message.content

@task(name="extract_keywords")
def extract_keywords(text: str) -> list:
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": "Extract 5 keywords. Return comma-separated."},
            {"role": "user", "content": text}
        ]
    )
    return response.choices[0].message.content.split(", ")

@workflow(name="process_document")
def process_document(document: str) -> dict:
    summary = summarize(document)
    keywords = extract_keywords(document)
    return {"summary": summary, "keywords": keywords}

# Run it
result = process_document("Your document text here...")

from anyway.sdk import Traceloop
from anyway.sdk.decorators import workflow, task
from anthropic import Anthropic

Traceloop.init(app_name="document-processor")
client = Anthropic()

@task(name="summarize")
def summarize(text: str) -> str:
    message = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        system="Summarize the text concisely.",
        messages=[{"role": "user", "content": text}]
    )
    return message.content[0].text

@task(name="extract_keywords")
def extract_keywords(text: str) -> list:
    message = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=256,
        system="Extract 5 keywords. Return comma-separated.",
        messages=[{"role": "user", "content": text}]
    )
    return message.content[0].text.split(", ")

@workflow(name="process_document")
def process_document(document: str) -> dict:
    summary = summarize(document)
    keywords = extract_keywords(document)
    return {"summary": summary, "keywords": keywords}

# Run it
result = process_document("Your document text here...")

Streaming

OpenAI
Anthropic

@task(name="stream_chat")
def stream_chat(prompt: str) -> str:
    stream = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}],
        stream=True
    )

    full_response = ""
    for chunk in stream:
        content = chunk.choices[0].delta.content or ""
        full_response += content
        print(content, end="", flush=True)

    return full_response

@task(name="stream_chat")
def stream_chat(prompt: str) -> str:
    full_response = ""

    with client.messages.stream(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}]
    ) as stream:
        for text in stream.text_stream:
            full_response += text
            print(text, end="", flush=True)

    return full_response

Tool Use / Function Calling

OpenAI
Anthropic

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get the current weather",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {"type": "string"}
                },
                "required": ["location"]
            }
        }
    }
]

@task(name="call_with_tools")
def call_with_tools(query: str):
    return client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": query}],
        tools=tools
    )

@task(name="execute_tool")
def execute_tool(tool_call):
    if tool_call.function.name == "get_weather":
        return {"temperature": "72°F", "condition": "sunny"}

@workflow(name="tool_workflow")
def tool_workflow(query: str):
    response = call_with_tools(query)
    if response.choices[0].message.tool_calls:
        for tool_call in response.choices[0].message.tool_calls:
            result = execute_tool(tool_call)
    return response

tools = [
    {
        "name": "get_weather",
        "description": "Get the current weather",
        "input_schema": {
            "type": "object",
            "properties": {
                "location": {"type": "string"}
            },
            "required": ["location"]
        }
    }
]

@task(name="call_with_tools")
def call_with_tools(query: str):
    return client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        tools=tools,
        messages=[{"role": "user", "content": query}]
    )

@task(name="execute_tool")
def execute_tool(tool_use):
    if tool_use.name == "get_weather":
        return {"temperature": "72°F", "condition": "sunny"}

@workflow(name="tool_workflow")
def tool_workflow(query: str):
    response = call_with_tools(query)
    for block in response.content:
        if block.type == "tool_use":
            result = execute_tool(block)
    return response

Async Support

Both decorators work seamlessly with async functions:

OpenAI
Anthropic

from openai import AsyncOpenAI

async_client = AsyncOpenAI()

@task(name="async_chat")
async def async_chat(prompt: str) -> str:
    response = await async_client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

@workflow(name="async_workflow")
async def async_workflow(questions: list[str]) -> list[str]:
    import asyncio
    tasks = [async_chat(q) for q in questions]
    return await asyncio.gather(*tasks)

from anthropic import AsyncAnthropic

async_client = AsyncAnthropic()

@task(name="async_chat")
async def async_chat(prompt: str) -> str:
    message = await async_client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}]
    )
    return message.content[0].text

@workflow(name="async_workflow")
async def async_workflow(questions: list[str]) -> list[str]:
    import asyncio
    tasks = [async_chat(q) for q in questions]
    return await asyncio.gather(*tasks)

View in Dashboard

After running your code, view traces in the Anyway Dashboard:

Navigate to Traces
Find your workflow trace
Expand to see nested task spans
View timing, inputs, and outputs for each operation

Next Steps

Configuration

Configure endpoints and authentication

Customer & Order Attribution

Link traces to customers and orders

Cost Tracking

Monitor your AI spend

​Overview

​Basic Setup

​The @workflow Decorator

​The @task Decorator

​Complete Example

​Streaming

​Tool Use / Function Calling

​Async Support

​View in Dashboard

​Next Steps