Burr Integration

Updated on Aug 14, 2026

Use this integration when your Burr application is implemented in Python.

Setup

1. Install SDK

bash

pip install progress-observability

2. Instrument your app

python

Observability.instrument(
    app_name=os.getenv("OBSERVABILITY_APP_NAME"),
    api_key=os.getenv("OBSERVABILITY_API_KEY")
)

3. Complete example

Note: Burr has no auto-instrumentation of its own actions for this OTel setup, so only explicit @workflow/@tool decorators create the parent-child span nesting needed for a structured trace tree.

python

import json
import os

from dotenv import load_dotenv
from burr.core import ApplicationBuilder, Result, State, action, default
from openai import OpenAI
from progress.observability import Observability, tool, workflow

load_dotenv()

Observability.instrument(
    app_name=os.getenv("OBSERVABILITY_APP_NAME"),
    api_key=os.getenv("OBSERVABILITY_API_KEY")
)

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
model = os.getenv("OPENAI_MODEL", "gpt-4o-mini")


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


# Use @tool to create a child span under the active @workflow span
@tool()
def get_weather(location: str) -> str:
    return f"The weather in {location} is cloudy with a high of 15°C."


def call_llm(messages: list, use_tools: bool = True):
    return client.chat.completions.create(
        model=model,
        messages=[{"role": "system", "content": "You are a helpful AI assistant."}, *messages],
        tools=TOOLS if use_tools else None,
        tool_choice="auto" if use_tools else None,
    ).choices[0].message


# Use @workflow to create the root span that groups all LLM and tool calls
@workflow(name="weather_agent")
def run_agent(messages: list) -> list:
    messages = list(messages)

    response = call_llm(messages)
    messages.append(response.model_dump(exclude_none=True))

    for tool_call in response.tool_calls or []:
        args = json.loads(tool_call.function.arguments)
        result = get_weather(**args)
        messages.append({"role": "tool", "tool_call_id": tool_call.id, "content": result})

    if response.tool_calls:
        final = call_llm(messages, use_tools=False)
        messages.append(final.model_dump(exclude_none=True))

    return messages


@action(reads=[], writes=["messages"])
def agent_step(state: State, payload: dict):
    messages = run_agent(payload["messages"])
    return {"messages": messages}, state.update(messages=messages)


agent_app = (
    ApplicationBuilder()
    .with_actions(agent_step=agent_step, result=Result("messages"))
    .with_transitions(("agent_step", "result", default))
    .with_entrypoint("agent_step")
    .with_state(messages=[])
    .build()
)

_, _, state = agent_app.run(
    halt_after=["result"],
    inputs={"payload": {"messages": [{"role": "user", "content": "What's the weather in Paris?"}]}},
)

print(state["messages"][-1]["content"])
Observability.shutdown()