Skip to content

symfonic.core.tools.streaming

streaming

Async-generator tool streaming support — Roadmap Item 7.

A tool authored as async def gen(...): yield ... carries semantics the LangGraph prebuilt ToolNode cannot directly consume: it calls tool.ainvoke(...) and expects a single awaitable result, not an async iterator. This module bridges the gap without introducing a new transport:

  1. wrap_async_gen_tool detects inspect.isasyncgenfunction on a StructuredTool's coroutine attribute and returns a replacement tool whose coroutine is a plain async def that drains the generator, dispatching every non-final yield as a custom event named tool_progress via langchain_core.callbacks.adispatch_custom_event. The final yield is returned as the tool result.
  2. prepare_tools_for_streaming applies the wrap helper to a list of tools, returning a list of identical length where async-gen tools are replaced and every other tool is passed through by identity (zero behaviour change for callers that do not opt in).

Why adispatch_custom_event is the right channel: LangGraph's astream_events(v2) already pipes such events to consumers as {"event": "on_custom_event", "name": ..., "data": ..., "run_id": ...} where run_id is the parent tool call. The transpiler (EventTranspiler._handle_tool_progress) then converts them to typed ToolProgressEvent items, so the new event flows through the same backpressure-aware iterator that already buffers ToolResultEvent.

prepare_tools_for_streaming 

prepare_tools_for_streaming(
    tools: list[BaseTool],
) -> list[BaseTool]

Apply wrap_async_gen_tool to each tool, preserving order.

Tools that are not async-generator StructuredTools are returned by identity, so callers can pass the result directly to langgraph.prebuilt.ToolNode with zero behaviour change for existing non-generator tools.

Source code in src/symfonic/core/tools/streaming.py 
150
151
152
153
154
155
156
157
158
def prepare_tools_for_streaming(tools: list[BaseTool]) -> list[BaseTool]:
    """Apply ``wrap_async_gen_tool`` to each tool, preserving order.

    Tools that are not async-generator ``StructuredTool``s are returned
    by identity, so callers can pass the result directly to
    ``langgraph.prebuilt.ToolNode`` with zero behaviour change for
    existing non-generator tools.
    """
    return [wrap_async_gen_tool(t) for t in tools]

wrap_async_gen_tool 

wrap_async_gen_tool(tool: BaseTool) -> BaseTool

Replace an async-generator tool's coroutine with a streaming wrapper.

If tool is not a StructuredTool carrying an async-generator coroutine, the original tool is returned by identity. This keeps the wrap helper safe to call indiscriminately from preset wiring (see prepare_tools_for_streaming).

The replacement tool is constructed via StructuredTool.from_function rather than mutating the original object so the input tool remains frozen — preserving Pydantic invariants on BaseTool subclasses.

Source code in src/symfonic/core/tools/streaming.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def wrap_async_gen_tool(tool: BaseTool) -> BaseTool:
    """Replace an async-generator tool's coroutine with a streaming wrapper.

    If ``tool`` is not a ``StructuredTool`` carrying an async-generator
    ``coroutine``, the original ``tool`` is returned by identity.  This
    keeps the wrap helper safe to call indiscriminately from preset
    wiring (see ``prepare_tools_for_streaming``).

    The replacement tool is constructed via
    ``StructuredTool.from_function`` rather than mutating the original
    object so the input tool remains frozen — preserving Pydantic
    invariants on ``BaseTool`` subclasses.
    """
    coroutine = _coroutine_attr(tool)
    if coroutine is None or not inspect.isasyncgenfunction(coroutine):
        return tool
    if not isinstance(tool, StructuredTool):
        # We can only re-issue via StructuredTool.from_function.  A
        # non-StructuredTool with an async-gen coroutine is unusual
        # enough to log and skip rather than guess at a constructor.
        logger.debug(
            "wrap_async_gen_tool: tool %r has async-gen coroutine but is "
            "not a StructuredTool; leaving unchanged.",
            getattr(tool, "name", tool),
        )
        return tool

    wrapped_coroutine = _build_wrapper(coroutine)
    return StructuredTool.from_function(
        coroutine=wrapped_coroutine,
        name=tool.name,
        description=tool.description,
        args_schema=getattr(tool, "args_schema", None),
        return_direct=getattr(tool, "return_direct", False),
        infer_schema=False,
    )