feat(agno): add workflow wrapper for instrumentation

[kausmeows]

What does this PR do?

• Adds a workflow wrapper for instrumenting Agno's Workflows.
• For the following kind of workflow config-

basic_workflow = Workflow(
    name="Basic Linear Workflow",
    description="Research -> Summarize -> Condition(Fact Check) -> Write Article",
    steps=[
        research_step,
        summarize_step,
        Condition(
            name="fact_check_condition",
            description="Check if fact-checking is needed",
            evaluator=needs_fact_checking,
            steps=[fact_check_step],
        ),
        write_article,
    ],
)

• We get the structure like this-

---

Note

Add Agno workflow/step/parallel instrumentation with context propagation and graph-node spans, plus examples and tests.

• Instrumentation (Agno):
  • Add wrappers for Workflow.run/arun, Step.execute/execute_stream/aexecute/aexecute_stream, and Parallel.execute/execute_stream/aexecute/aexecute_stream with context propagation, graph node attributes, and streaming support.
  • Extract shared helpers to openinference.instrumentation.agno.utils and refactor _RunWrapper to use them; minor adjustments to streaming output handling.
• Examples:
  • Add workflow samples: workflow_with_condition_step.py, workflow_with_custom_function_step.py, workflow_with_parallel_step.py, workflow_with_structure_io.py.
  • Update examples/requirements.txt (upgrade agno, add arize-phoenix, openai).
• Tests:
  • Add test_workflow_instrumentation.py covering basic, multi-step, async, conditional, and graph hierarchy cases; minor fix in existing tests.
• Project config:
  • Add fastapi to test dependencies.

^{Written by Cursor Bugbot for commit edd117d. This will update automatically on new commits. Configure here.}

[cursor]

Bug: Non-streaming iterables incorrectly detected as streams

The streaming detection uses hasattr(result, "__iter__") which matches any iterable including lists, tuples, dicts, and sets. While it excludes strings and bytes, it doesn't exclude other non-streaming collections. If a step returns a list or dict, it will be incorrectly treated as a stream, causing the wrong instrumentation path and potentially incorrect span attributes. The check should use isinstance(result, Iterator) from the typing module to properly detect actual iterators.

 

[cursor]

Bug: Unchecked method call may raise AttributeError

The _extract_output function assumes that any response without a content attribute has a model_dump_json method (Pydantic models). If a response object has neither attribute, calling response.model_dump_json() raises an AttributeError. Since this instrumentation code wraps user workflows, unexpected response types could crash the instrumentation and potentially disrupt the user's workflow execution. The function also has unreachable code at line 101 since all prior branches return.

 

[cursor]

Bug: Async methods missing error handling for span cleanup

The arun methods in _WorkflowWrapper, _StepWrapper, and aexecute in _ParallelWrapper lack try/except/finally around the initial wrapped(*args, **kwargs) call. Unlike their sync counterparts which have proper exception handling, these async methods start a span but if wrapped() raises a synchronous exception before returning a coroutine, the span is never ended and the error is not recorded. This causes orphaned spans in the tracing system.

Additional Locations (2)

• python/instrumentation/openinference-instrumentation-agno/src/openinference/instrumentation/agno/_workflow_wrapper.py#L646-L656
• python/instrumentation/openinference-instrumentation-agno/src/openinference/instrumentation/agno/_workflow_wrapper.py#L934-L951