bat.prebuilt.call_agent_node

CallAgentNode Objects

class CallAgentNode(PrebuiltWorkflow)

CallAgentNode implements agent-to-agent communication using the A2A protocol.

This workflow abstracts streaming communication between agents as an internal mini-graph. It handles the complete lifecycle of calling another agent, consuming its streamed responses, and managing state updates.

The internal mini-graph flow is: START → call_agent → consume_stream → router → (consume_stream | cleanup) → END

The loop continues consuming streamed responses from the target agent until either:

• The target agent completes its task
• The target agent requests user input
• An error occurs

Example

from typing import Optional

from bat.agent import AgentConfig, AgentGraph, AgentState
from bat.prebuilt import CallAgentNode

class MyAgentState(AgentState):
    agent_input_text: str
    agent_output_text: str
    agent_status: str
    agent_response_status: str
    agent_input_required: bool = False
    agent_response_content: Optional[str] = None
    ...

class MyAgentGraph(AgentGraph):
    def setup(self, config: AgentConfig) -> None:
        self.call_agent_node = CallAgentNode(
            config=config,
            StateType=MyAgentState,
            loop_name="domain_agent_loop",
            agent_name="SMO Agent",
            input_key="agent_input_text",
            output_key="agent_output_text",
            status_key="agent_status",
            agent_input_required_key="agent_input_required",
            agent_response_status_key="agent_response_status",
            agent_response_content_key="agent_response_content",
            build_message=self.build_message,
        )
        ...
        self.graph_builder.add_node(
            "call_agent_node",
            self.call_agent_node.as_runnable()
        )

__init__

def __init__(config: AgentConfig,
             StateType: Type[AgentState],
             loop_name: str,
             agent_name: str,
             build_message: Callable[[RunnableConfig, str], Message],
             *,
             input_key: str = "input",
             output_key: str = "output",
             status_key: str = "status",
             agent_input_required_key: str = "agent_input",
             agent_response_status_key: str = "agent_response_status",
             agent_response_content_key: str = "agent_response_content",
             recursion_limit: int = 50) -> None

Initialize the CallAgentNode workflow with the given configuration and parameters.

Arguments:

• config AgentConfig - Configuration for the agent.
• StateType Type[AgentState] - The AgentState schema used in the loop.
• loop_name str - The name of this workflow loop (e.g., "domain_agent_loop").
• agent_name str - The name of the target agent to call (e.g., "SMO Agent"). The agent card will be retrieved from the configuration using this name.
• build_message Callable[[RunnableConfig, str], Message] - Callback function to build the request message from the input text. Should accept a RunnableConfig and a string, and return an A2A Message object.
• input_key str, optional - A key pointing to a string in the state. Defaults to "input". The value at this key is used as input to send to the target agent.
• output_key str, optional - A key pointing to a string in the state. Defaults to "output". The value at this key is updated with responses from the target agent.
• status_key str, optional - A key pointing to a string in the state. Defaults to "status". The value at this key is updated with the overall status of the communication. Useful to display the current operation to the user.
• agent_input_required_key str, optional - A key pointing to a bool in the state. Defaults to "agent_input". The value at this key is set to True when the target agent requests user input.
• agent_response_status_key str, optional - A key pointing to a string in the state. Defaults to "agent_response_status". The value at this key is updated with the status from the target agent.
• agent_response_content_key str, optional - A key pointing to a string in the state. Defaults to "agent_response_content". The value at this key is updated with the content from the target agent.
• recursion_limit int, optional - Maximum recursion depth for nested calls. Defaults to 50. This prevents infinite loops in agent-to-agent communication.

consume_agent_stream

async def consume_agent_stream(
        agent_card: AgentCard,
        message: Message) -> AsyncIterable[StreamResponse]

Consume the agent stream from another A2A agent. The following operations are performed:

1. Creates an A2A client configured for streaming (120s timeout)
2. Opens a CLIENT span and propagates its trace context to the called agent via the message metadata (W3C traceparent)
3. Sends the request message to the target agent
4. Yields each stream item (events or messages)
5. Handles errors and ensures the span is closed

Token usage and tool calls of the called agent are captured through its own OpenTelemetry spans (correlated to this call by the shared trace_id), not extracted from the A2A messages here.

Arguments:

• agent_card AgentCard - The agent card of the target agent, containing connection details and capabilities.
• message Message - The A2A message to send to the agent.

Yields:

• StreamResponse - Stream items from the agent, including status updates, artifacts, and final messages.

Raises:

• Exception - If the streaming connection fails or encounters an error.

bat.prebuilt.prebuilt_workflow

PrebuiltWorkflow Objects

class PrebuiltWorkflow(ABC)

Abstract base class for prebuilt workflows. Extend this class to implement custom workflows.

__init__

def __init__(config: AgentConfig, StateType: Type[AgentState], *args,
             **kwargs)

Initialize the workflow: set up the graph with _setup and compile it.

The graph is compiled without a checkpointer: run as a node of an AgentGraph, it uses the checkpointer of that graph.

Arguments:

• config AgentConfig - The agent configuration.
• StateType Type[AgentState] - The AgentState schema of the workflow. *args, **kwargs: Forwarded to _setup.

StateType

@property
def StateType() -> Type[AgentState]

Get the AgentState type used in the workflow.

Returns:

• Type[AgentState] - The AgentState type.

graph_builder

@property
def graph_builder() -> StateGraph

Get the state graph builder.

Returns:

• StateGraph - The state graph builder.

graph

@property
def graph() -> CompiledStateGraph

Get the compiled state graph.

Returns:

• CompiledStateGraph - The compiled state graph.

agent_config

@property
def agent_config() -> AgentConfig

Get the AgentConfig used in the workflow.

Returns:

• AgentConfig - The AgentConfig.

as_runnable

def as_runnable() -> RunnableGenerator

Returns a RunnableGenerator for the ReActLoop, wrapping the astream method. Allows the ReActLoop to be used as a node (subgraph) in a larger graph.

bat.prebuilt.react_loop

ReActLoop Objects

class ReActLoop(PrebuiltWorkflow)

ReActLoop implements a ReAct-style loop using a ChatModelClient with associated tools.

The available tools must be registered in the ChatModelClient used to create the ReActLoop. The loop continues until the chat model produces a final answer without any tool calls.

Args

config (AgentConfig): Configuration for the agent.
StateType (Type[AgentState]): The AgentState schema used in the loop.
loop_name (str): The name of the loop.
chat_model_client (ChatModelClient): The chat model client to use.
input_key (str, optional): A key pointing to a string or HumanMessage
    in the state. Defaults to "input".
    The value at this key is used as input to the ChatModelClient on
    the first call of the loop.
output_key (str, optional): A key pointing to a string in the state.
    Defaults to "output".
    The value at this key is set to the final answer from the
    ChatModelClient when the loop completes.
messages_key (str, optional): A key pointing to a List[BaseMessage]
    in the state. Defaults to None.
    If provided, the value at this key is used as the conversation
    history for the ChatModelClient.
    The conversation history is updated **in-place** with each response
    from the ChatModelClient.
status_key (str, optional): A key pointing to a string in the state.
    Defaults to None.
    If provided, the value at this key is updated with the current
    status of the loop.
    Useful to beautify the streamed output of the loop.
tool_call_policy ("parallel" | "sequential", optional): How a turn
    asking for several tools is executed. Defaults to "parallel",
    which is the behaviour of every earlier release: the ToolNode runs
    them together on one snapshot of the state.
    "sequential" executes them one at a time, each on the state the
    previous one left. Required by any loop whose tools write state
    through ``Command``: siblings in one step cannot see each other,
    and LangGraph rejects two writes to the same key in a single step.
    It costs no extra model call — the request is not split, only its
    execution.

Example

from typing import List, Optional

from bat.agent import AgentConfig, AgentGraph, AgentState
from bat.chat_model_client import ChatModelClient
from bat.prebuilt import ReActLoop
from langchain.tools import tool
from langchain_core.messages import BaseMessage

class RNGClient(ChatModelClient):
    ...

class RNGAgentState(AgentState):
    user_input: str
    model_output: str
    conversation_history: List[BaseMessage] = []
    agent_status: Optional[str] = None
    ...

@tool
def generate_random_numbers(min: int, max: int, n: int) -> str:
    """Generate n random numbers between min and max (inclusive)."""
    ...

class RNGAgentGraph(AgentGraph):
    def setup(self, config: AgentConfig) -> None:
        self.rng_client = RNGClient(tools=[generate_random_numbers])
        self.rng_loop = ReActLoop(
            config=config,
            StateType=RNGAgentState,
            loop_name="rng_loop",
            chat_model_client=self.rng_client,
            input_key="user_input",
            output_key="model_output",
            messages_key="conversation_history",
            status_key="agent_status",
        )
        ...
        self.graph_builder.add_node("rng_loop", self.rng_loop.as_runnable())

__init__

def __init__(
        config: AgentConfig,
        StateType: Type[AgentState],
        loop_name: str,
        chat_model_client: ChatModelClient,
        input_key: str = "input",
        output_key: str = "output",
        messages_key: Optional[str] = None,
        status_key: Optional[str] = None,
        tool_call_policy: Literal["parallel",
                                  "sequential"] = "parallel") -> None

Initialize the ReActLoop with the given configuration and parameters.

Arguments:

• config AgentConfig - Configuration for the agent.
• StateType Type[AgentState] - The AgentState schema used in the loop.
• loop_name str - The name of the loop.
• chat_model_client ChatModelClient - The chat model client to use.
• input_key str, optional - A key pointing to a string or HumanMessage in the state. Defaults to "input". The value at this key is used as input to the ChatModelClient on the first call of the loop.
• output_key str, optional - A key pointing to a string in the state. Defaults to "output". The value at this key is set to the final answer from the ChatModelClient when the loop completes.
• messages_key str, optional - A key pointing to a List[BaseMessage] in the state. Defaults to None. If provided, the value at this key is used as the conversation history for the ChatModelClient. The conversation history is updated in-place with each response from the ChatModelClient.
• status_key str, optional - A key pointing to a string in the state. Defaults to None. If provided, the value at this key is updated with the current status of the loop. Useful to beautify the streamed output of the loop.
• tool_call_policy "parallel" | "sequential", optional - How a turn asking for several tools is executed. Defaults to "parallel": the ToolNode runs them together on one snapshot of the state. "sequential" executes them one at a time, each on the state the previous one left. Required by any loop whose tools write state through Command.