From b62623c1a73ff0a8317861a34b8ed466f051069a Mon Sep 17 00:00:00 2001 From: aifanatic Date: Tue, 3 Nov 2026 23:30:30 -0600 Subject: [PATCH] Python: Fix custom client, fan-in, and executor docstring examples Co-Authored-By: Claude Opus 5.5 --- .../packages/core/agent_framework/_clients.py | 36 ++++++++++++------- .../agent_framework/_workflows/_executor.py | 6 ++++ .../_workflows/_workflow_builder.py | 15 +++++++- 3 files changed, 43 insertions(+), 14 deletions(-) diff --git a/python/packages/core/agent_framework/_clients.py b/python/packages/core/agent_framework/_clients.py index ed49727411c..f61345c43c3 100644 --- a/python/packages/core/agent_framework/_clients.py +++ b/python/packages/core/agent_framework/_clients.py @@ -244,26 +244,36 @@ class BaseChatClient(SerializationMixin, ABC, Generic[OptionsCoT]): Examples: .. code-block:: python - from agent_framework import BaseChatClient, ChatResponse, Message - from collections.abc import AsyncIterable + from agent_framework import ( + BaseChatClient, + ChatResponse, + ChatResponseUpdate, + Content, + Message, + ResponseStream, + ) class CustomChatClient(BaseChatClient): - async def _inner_get_response(self, *, messages, stream, options, **kwargs): + # Not ``async``: return an awaitable for a single response, or a + # ResponseStream when ``stream`` is True. + def _inner_get_response(self, *, messages, stream, options, **kwargs): + response = ChatResponse( + messages=[Message(role="assistant", contents=["Hello!"])], + response_id="custom-response", + ) if stream: # Streaming implementation - from agent_framework import ChatResponseUpdate - async def _stream(): - yield ChatResponseUpdate(role="assistant", contents=[{"type": "text", "text": "Hello!"}]) + yield ChatResponseUpdate(role="assistant", contents=[Content.from_text("Hello!")]) - return _stream() - else: - # Non-streaming implementation - return ChatResponse( - messages=[Message(role="assistant", contents=["Hello!"])], - response_id="custom-response", - ) + return ResponseStream(_stream(), finalizer=lambda updates: response) + + # Non-streaming implementation + async def _get_response(): + return response + + return _get_response() # Create an instance of your custom client diff --git a/python/packages/core/agent_framework/_workflows/_executor.py b/python/packages/core/agent_framework/_workflows/_executor.py index 7c1a6ffd524..cb934ff2568 100644 --- a/python/packages/core/agent_framework/_workflows/_executor.py +++ b/python/packages/core/agent_framework/_workflows/_executor.py @@ -53,6 +53,8 @@ class MyExecutor(Executor): @handler async def handle_string(self, message: str, ctx: WorkflowContext) -> None: # This executor can handle 'str' input types + ... + Access via the `input_types` property. ### Output Types @@ -64,6 +66,8 @@ class MyExecutor(Executor): @handler async def handle_data(self, message: str, ctx: WorkflowContext[int | bool]) -> None: # This executor can send 'int' or 'bool' messages + ... + Access via the `output_types` property. ### Workflow Output Types @@ -75,6 +79,8 @@ class MyExecutor(Executor): @handler async def process(self, message: str, ctx: WorkflowContext[int, str]) -> None: # Can send 'int' messages AND yield 'str' workflow outputs + ... + Access via the `workflow_output_types` property. ## Handler Discovery diff --git a/python/packages/core/agent_framework/_workflows/_workflow_builder.py b/python/packages/core/agent_framework/_workflows/_workflow_builder.py index 19b5eb48799..eb6124b5ece 100644 --- a/python/packages/core/agent_framework/_workflows/_workflow_builder.py +++ b/python/packages/core/agent_framework/_workflows/_workflow_builder.py @@ -533,6 +533,12 @@ def add_fan_in_edges( from agent_framework import Executor, WorkflowBuilder, WorkflowContext, handler + class Dispatcher(Executor): + @handler + async def dispatch(self, seed: int, ctx: WorkflowContext[int]) -> None: + await ctx.send_message(seed) + + class Producer(Executor): @handler async def produce(self, seed: int, ctx: WorkflowContext[str]) -> None: @@ -546,11 +552,18 @@ async def aggregate(self, results: list[str], ctx: WorkflowContext[Never, str]) await ctx.yield_output(f"Combined: {combined}") + dispatcher = Dispatcher(id="dispatcher") prod_1 = Producer(id="prod_1") prod_2 = Producer(id="prod_2") agg = Aggregator(id="agg") - workflow = WorkflowBuilder(start_executor=prod_1).add_fan_in_edges([prod_1, prod_2], agg).build() + # Every fan-in source must be reachable, so the dispatcher fans out to both producers. + workflow = ( + WorkflowBuilder(start_executor=dispatcher) + .add_fan_out_edges(dispatcher, [prod_1, prod_2]) + .add_fan_in_edges([prod_1, prod_2], agg) + .build() + ) """ source_execs = [self._maybe_wrap_agent(s) for s in sources] target_exec = self._maybe_wrap_agent(target)