Example: Microsoft Agent Framework

Prerequisites: 02-tracing-observations.md and 05-tracing-frameworks.md.

An order-processing workflow. The orchestrator inspects order priority and routes to either an express or standard fulfillment executor. The fulfillment executor processes the order and returns a shipping result.

"""
Tracing annotations:
- setup_tracing() with a global InputFilter that removes WorkflowContext from
  all observations (it serializes noisy internal framework state).
- @step on each executor handler creates one observation per handler invocation.
- as_type="agent" marks executors that make routing or generation decisions.
- event() marks the decision point where high-priority orders are detected.
- update_current_step() attaches runtime routing/result metadata.
- exclude_args=["debug_info"] demonstrates per-step filtering for an argument
  not covered by the global filter.
- as_type="tool" marks the shipping API call with the Langfuse tool type.
"""

from agent_framework import Executor, Workflow, WorkflowBuilder, WorkflowContext, handler
from pydantic import BaseModel

from semantic_tracing import InputFilter, event, setup_tracing, span, step, update_current_step


# --- models ---

class OrderRequest(BaseModel):
    id: str
    priority: str
    items: list[str]

class FulfillRequest(BaseModel):
    order: OrderRequest
    lane: str

class ShippingResult(BaseModel):
    status: str
    tracking_id: str


# --- tracing init (call once at process startup) ---

setup_tracing(
    service_name="order_workflow",
    input_filter=InputFilter(
        exclude_arg_names=["ctx"],
        exclude_arg_types=["WorkflowContext"],
    ),
)


# --- orchestrator executor ---

class OrchestratorExecutor(Executor):
    def __init__(self):
        super().__init__(id="orchestrator")

    @step("orchestrate", as_type="agent", exclude_args=["debug_info"])
    @handler
    async def handle(
        self, request: OrderRequest, ctx: WorkflowContext, debug_info: dict | None = None
    ) -> None:
        if request.priority == "high":
            target = "express"
            event("high_priority_detected", metadata={"order_id": request.id})
        else:
            target = "standard"

        update_current_step(metadata={"routed_to": target})
        await ctx.send(FulfillRequest(order=request, lane=target))


# --- fulfillment executor ---

class FulfillExecutor(Executor):
    def __init__(self):
        super().__init__(id="fulfill")

    @step("fulfill_order", as_type="agent")
    @handler
    async def handle(self, request: FulfillRequest, ctx: WorkflowContext) -> None:
        with span("call_shipping_api", as_type="tool") as s:
            result = ship_order(request.order, request.lane)
            s.update(output={"tracking_id": result.tracking_id})

        update_current_step(
            output={"status": result.status, "tracking": result.tracking_id},
            metadata={"lane": request.lane},
        )
        await ctx.send(result)


# --- workflow assembly and execution ---

def build_workflow() -> Workflow:
    orchestrator = OrchestratorExecutor()
    fulfill = FulfillExecutor()
    builder = WorkflowBuilder(start_executor=orchestrator)
    builder.add_edge(source=orchestrator, target=fulfill)
    return builder.build()


async def main():
    workflow = build_workflow()
    request = OrderRequest(id="ORD-42", priority="high", items=["widget-a"])
    await workflow.run(request)

Resulting Langfuse trace:

trace (order_workflow)
├── orchestrate (agent)
│     input: {"request": {"id": "ORD-42", "priority": "high", "items": [...]}}
│     metadata: {"routed_to": "express"}
│     └── high_priority_detected (event)
│           metadata: {"order_id": "ORD-42"}
└── fulfill_order (agent)
      input: {"request": {"order": {...}, "lane": "express"}}
      output: {"status": "shipped", "tracking": "TRK-99"}
      metadata: {"lane": "express"}
      └── call_shipping_api (tool)
            output: {"tracking_id": "TRK-99"}