Documentation Index
Fetch the complete documentation index at: https://docs.anyway.sh/llms.txt
Use this file to discover all available pages before exploring further.
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}
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
Complete Example
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
@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
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:
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
