bat.agent.application

AgentApplication Objects

class AgentApplication()

Agent Application based on Starlette. This class sets up an agent application that serves the A2A protocol.

Configuration: Read from the YAML file at CONFIG_PATH (default ./config.yaml); see :class:AgentConfig for its fields.

Environment variables read here:
- ``CONFIG_PATH``: path of the config file.
- ``AGENT_CARD_PATH``: agent card path, used when config.yaml has no
  ``agent_card`` (default ``./agent.json``).
- ``AGENT_CARD_DISPLAY``: ``1`` (default) prints the agent card at
  startup.

Attributes

agent_card (AgentCard): The agent card containing metadata about the
    agent.
agent_graph (AgentGraph): The agent graph that defines the agent's
    behavior and capabilities.

Example

    from bat.agent import AgentApplication

    agent = AgentApplication(
        AgentGraphType=MyAgentGraph,
        AgentStateType=MyAgentState,
    )
    agent.run()

__init__

def __init__(AgentGraphType: Type[AgentGraph],
             AgentStateType: Type[AgentState],
             telemetry_privacy_floor: TelemetryPrivacyLevel = "none")

Initialize the AgentApplication with the agent's graph and state.

Everything else is read from config.yaml (see :class:AgentConfig).

Arguments:

• AgentGraphType Type[AgentGraph] - The class to use to instantiate the agent graph.
• AgentStateType Type[AgentState] - The class to use to instantiate the agent state.
• telemetry_privacy_floor TelemetryPrivacyLevel - The lowest telemetry privacy level this agent allows -- "none", "content", "names" or "full". Defaults to "none", which leaves telemetry.privacy in config.yaml to decide.

load_agent_card

def load_agent_card(agent_card_path: str) -> AgentCard

Load the Agent Card from a JSON file.

Arguments:

• agent_card_path str - The path to the Agent Card JSON file.

Returns:

• AgentCard - The loaded Agent Card.

Raises:

• Exception - For general errors during loading.
• EnvironmentError - If endpoint.url is not set in config.yaml.
• FileNotFoundError - If the agent card file does not exist.
• ValidationError - If the agent card JSON is invalid.

agent_graph

@property
def agent_graph() -> AgentGraph

Get the agent graph.

agent_card

@property
def agent_card() -> AgentCard

Get the agent card.

run

def run() -> None

Run the agent application, serving the A2A protocol.

bat.agent.state

AgentTaskStatus Objects

class AgentTaskStatus(IntEnum)

AgentTaskStatus is an enum type matching the A2A-SDK TaskState. The need for this redefinition is due to: - Incompatibility between PydanticV2 and Protobuf. - Not all the A2A-SDK TaskState are currently supported by BAT-ADK.

Among those defined in the A2A-SDK, BAT-ADK currently supports:

• AGENT_TASK_STATUS_WORKING -> TASK_STATE_WORKING
• AGENT_TASK_STATUS_INPUT_REQUIRED -> TASK_STATE_INPUT_REQUIRED
• AGENT_TASK_STATUS_COMPLETED -> TASK_STATE_COMPLETED
• AGENT_TASK_STATUS_FAILED -> TASK_STATE_FAILED

Internally, the following A2A-SDK task states are handles:

• TASK_STATE_SUBMITTED

The A2A-SDK defines the following additional task states:

• TASK_STATE_UNSPECIFIED
• TASK_STATE_CANCELED
• TASK_STATE_REJECTED
• TASK_STATE_AUTH_REQUIRED

AgentTaskResult Objects

class AgentTaskResult(BaseModel)

Result of an agent invocation.

Attributes

task_status (AgentTaskStatus): The status of the agent task.
content (str): The content of the agent's response or message.

Attributes meaning

task_status	content
TASK_STATE_WORKING	Description of task progress
TASK_STATE_INPUT_REQUIRED	Description of required input or context
TASK_STATE_COMPLETED	Final response of the agent
TASK_STATE_FAILED	Error message explaining the error

__init__

def __init__(task_status: AgentTaskStatus | str, content: str)

Initialize AgentTaskResult with task status and content.

Arguments:

• task_status AgentTaskStatus | str - The status of the agent task. Can be provided as an AgentTaskStatus enum or a string (deprecated).
• content str - The content of the agent's response or message.

from_send_message_stream

@classmethod
def from_send_message_stream(cls, chunk: StreamResponse) -> Self

Instantiate an AgentTaskResult from an A2A StreamResponse object. Handles different types of StreamResponse as follows:

1. Message: Direct message response → (completed, content)
2. TaskArtifactUpdateEvent: Artifact update → (completed, content)
3. TaskStatusUpdateEvent: Status update → (TaskStatusUpdateEvent.state, content)
4. Task: Not supported yet → (working, undefined content)

The content is extracted with the get_stream_response_text utility function of A2A-SDK.

Arguments:

• chunk StreamResponse - Stream chunk generated by A2A-SDK send_message.

Returns:

• AgentTaskResult - as reported above.

requires_input

def requires_input() -> bool

Returns true when task_status indicates that user input is required.

AgentState Objects

class AgentState(BaseModel, ABC)

Abstract Pydantic model from which agent's state classes should inherit.

This class combines Pydantic's model validation with abstract state management requirements for agent operations. Subclasses should define concrete state models while implementing the required abstract methods.

Attributes

bat_extra (Dict[str, Any]): A dictionary for storing extra state
    information.
    The user should not modify this directly, as it is used internally
    by the SDK.
bat_buffer (List): A list used as a buffer for intermediate state data.
    The user should not modify this directly, as it is used internally
    by the SDK.

Methods

from_query (**abstract**): Factory method to create an agent state from
    an initial query
to_task_result (**abstract**): Convert current state to a
    `AgentTaskResult` object
update_after_checkpoint_restore: Refresh state after checkpoint
    restoration
is_waiting_for_human_input: Check if agent requires human input

Example

from bat.agent import AgentState, AgentTaskResult, AgentTaskStatus
from typing import List, Optional, Self
from typing_extensions import override

class MyAgentState(AgentState):
    user_inputs: List[str] = []
    assistant_outputs: List[str] = []
    question: str = ""
    answer: Optional[str] = None

    @classmethod
    def from_query(cls, query: str) -> Self:
        return cls(
            user_inputs=[query],
            question=query,
        )

    @override
    def update_after_checkpoint_restore(self, query: str) -> None:
        self.user_inputs.append(query)
        self.question = query

    @override
    def to_task_result(self) -> AgentTaskResult:
        if self.answer is None:
            return AgentTaskResult(
                task_status=AgentTaskStatus.AGENT_TASK_STATUS_WORKING,
                content="Processing your request..."
            )
        return AgentTaskResult(
            task_status=AgentTaskStatus.AGENT_TASK_STATUS_COMPLETED,
            content=self.answer
        )

from_query

@classmethod
@abstractmethod
def from_query(cls, query: str) -> Self

Instantiate agent state from initial query.

Factory method called by the execution framework to create a new state instance. Alternative to direct initialization, allowing state-specific construction logic.

Arguments:

• query - Initial user query to bootstrap agent state

Returns:

• Self - Fully initialized agent state instance

update_after_checkpoint_restore

def update_after_checkpoint_restore(query: str) -> None

Update state with new query after checkpoint restoration.

Called by the SDK when a message arrives in a conversation with a saved checkpoint, before is_waiting_for_human_input.

By default, the state starts over from from_query(query), so the agent answers as if there were no checkpoint. While the agent is waiting for human input, the restored state is kept for the resume. Override to keep data across the messages of a conversation.

Arguments:

• query - New query to execute with the restored state

to_task_result

@abstractmethod
def to_task_result() -> AgentTaskResult

Convert current state to a task result object.

Used to yield execution results during graph processing. This method defines how the agent's internal state translates to external-facing task results.

It is called on every state while the graph runs, including states inside prebuilt workflows. WORKING results are sent as progress. The answer is the result of the graph's final state, so this method may return COMPLETED as soon as the answer is set, even if later nodes change it.

Returns:

• AgentTaskResult - Task result representation of current state

is_waiting_for_human_input

def is_waiting_for_human_input() -> bool

Check if agent is blocked waiting for human input.

Default implementation returns False. Override in subclasses to implement human-in-the-loop pausing behavior.

Returns:

• bool - True if agent requires human input to proceed, False otherwise

bat.agent.config

MCPServerConfig Objects

class MCPServerConfig(BaseModel)

Configuration for an MCP server connection.

Attributes:

• name str - The name of the MCP server.

• url str - The URL of the MCP server.

• required bool - Whether the MCP server is required to be reachable. Defaults to True. When set to False, connection failures will be treated as if the server is available but always returns empty responses (e.g., no tools).

• timeout int - The timeout in seconds for connecting to the MCP server. Defaults to 60.

• Note - the required attribute is not well supported yet.

RemoteAgentConfig Objects

class RemoteAgentConfig(BaseModel)

Configuration for a remote A2A agent connection.

Attributes:

• name str - The name of the remote agent.

• url str - The URL of the remote agent.

• required bool - Whether the remote agent is required to be reachable. Defaults to True. When set to False, connection failures will be treated by ignoring the agent, i.e. no AgentCard will be returned.

• timeout int - The timeout in seconds for connecting to the remote agent. Defaults to 60.

• Note - the required attribute is not well supported yet.

EndpointConfig Objects

class EndpointConfig(BaseModel)

Where the agent is served (base URL and port).

Read directly by AgentApplication (not via environment variables).

A port embedded directly in url (e.g. http://host:8080) is lifted out into port so a single value is authoritative for both binding the server and advertising the agent card. Specifying the port in both places is allowed only if they agree; a mismatch is a configuration error.

Attributes:

• url Optional[str] - Base URL the agent is hosted at (used to build the agent card interface URL). Stored without a port.
• port Optional[int] - A2A server port. Defaults to 9900.

ModelConfig Objects

class ModelConfig(BaseModel)

Chat model selection.

Only provider, name and base_url may be overridden by an environment variable: MODEL / MODEL_PROVIDER / BASE_URL take precedence when set (see :meth:bat.chat_model_client.ChatModelClientConfig.load).

Attributes:

• provider Optional[str] - Model provider, e.g. openai.
• name Optional[str] - Model name, e.g. gpt-4.1-mini.
• base_url Optional[str] - Optional base URL for the provider; needed by local providers such as ollama.
• reasoning_effort Optional[str] - How much the model reasons before answering, e.g. low. Values depend on the provider.
• service_tier Optional[str] - Processing tier for the requests, e.g. flex. Values depend on the provider.

OutputConfig Objects

class OutputConfig(BaseModel)

A single telemetry export destination.

Attributes:

• type str - local (JSONL file), remote (OTLP/Phoenix) or console (stdout).
• file_path Optional[str] - Target file for local (defaults to spans.jsonl).
• endpoint Optional[str] - OTLP collector base URL for remote (defaults to http://localhost:6006).

TelemetrySettings Objects

class TelemetrySettings(BaseModel)

OpenTelemetry settings (the enabling switch and exporter list).

Named TelemetrySettings to avoid colliding with bat.telemetry.config.TelemetryConfig. AgentApplication reads these directly and feeds them to TelemetryConfig.from_settings.

Spans are fanned out to every entry in output, e.g.::

telemetry: output:

• type: local file_path: spans.jsonl
• type: remote endpoint: http://localhost:6006

Attributes:

• service_name Optional[str] - service.name; defaults to the agent card name.

• project_name Optional[str] - OpenInference/Phoenix project to file traces under (the openinference.project.name resource attribute). When unset, Phoenix uses the default project. This is distinct from service_name (which labels spans within a project). Agents that share a distributed trace must use the same project or the trace fragments across projects.

• privacy TelemetryPrivacy - How much of the agent's internals may leave the process. An ordered ladder, each level a superset of the one below::

  none # export everything (default) content # redact prompts, messages, completions,

  invocation parameters, and every tool's

  description, parameter schema and call arguments

  names # also redact span (LangGraph node) names full # also redact tool names

  Values replaced with __REDACTED__ never leave the process. Token counts, span kinds, hierarchy and timing survive at every level, so usage/cost tracking keeps working. full empties the eval engine's tool-call metrics, which key off the tool name. Written as the level name or its ordinal; an unknown value is a validation error rather than a silent fallback to none.

• output List[OutputConfig] - One entry per active destination.

AgentConfig Objects

class AgentConfig(BaseModel)

Agent Configuration, including MCP servers and remote agents.

Attributes

checkpoints (bool): Keep each conversation's state in memory between
    messages. Defaults to False.
agent_card (Optional[str]): Path to the agent card JSON file. When
    unset it defaults to ``./agent.json`` (see ``AgentApplication``).
endpoint (Optional[EndpointConfig]): Where the agent is served;
    ``endpoint.url`` is required by ``AgentApplication``.
model (Optional[ModelConfig]): Chat model selection and settings.
telemetry (Optional[TelemetrySettings]): OpenTelemetry settings.
mcp_servers (List[MCPServerConfig]): List of MCP server
    configurations (``mcp-servers`` in YAML).
remote_agents (List[RemoteAgentConfig]): List of remote agent
    configurations (``remote-agents`` in YAML).

Methods

is_remote_agent_required(agent_name: str) -> bool:
    Check if a remote agent is marked as required in the configuration.
is_mcp_server_required(server_name: str) -> bool:
    Check if an MCP server is marked as required in the configuration.
load(path: str = 'config.yaml') -> AgentConfig:
    Load the agent configuration from a YAML file.
get_tools(mcp_server_names: List[str]) -> List[BaseTool]:
    Retrieve tools from the specified MCP servers.
get_agent_cards(agent_names: List[str]) -> List[AgentCard]:
    Retrieve agent cards from the specified remote agents.

list_mcp_servers_names

def list_mcp_servers_names() -> List[str]

List the names of all configured MCP servers.

Returns:

• List[str] - List of MCP server names.

list_remote_agents_names

def list_remote_agents_names() -> List[str]

List the names of all configured remote agents.

Returns:

• List[str] - List of remote agent names.

is_remote_agent_required

def is_remote_agent_required(agent_name: str) -> bool

Check if a remote agent is marked as required in the configuration.

Arguments:

• agent_name str - The name of the remote agent.

Returns:

• bool - True if the agent is required, False otherwise (or if the agent is not in the configuration).

is_mcp_server_required

def is_mcp_server_required(server_name: str) -> bool

Check if an MCP server is marked as required in the configuration.

Arguments:

• server_name str - The name of the MCP server.

Returns:

• bool - True if the server is required, False otherwise (or if the server is not in the configuration).

load

@classmethod
def load(cls, path: str = "config.yaml") -> Self

Load the agent configuration from a YAML file. If the YAML file is not found, an empty configuration is used.

Environment variables referenced as $VAR or ${VAR} are expanded in the file before parsing, so values can be templated from the environment (e.g. port: ${PORT}). Unset variables are left literal.

Arguments:

• path str - The path to the configuration YAML file.

Returns:

• AgentConfig - The loaded agent configuration or an empty configuration if the file is not found.

Raises:

• ValueError - If the configuration file cannot be loaded or validated.

get_mcp_server_connection

def get_mcp_server_connection(server_name: str) -> MCPConnection | None

Get the MCP connection for a given MCP server.

Arguments:

• server_name str - The name of the MCP server.

Returns:

MCPConnection | None: The MCP connection object if the MCP server is found, otherwise None

get_a2a_agent_connection

def get_a2a_agent_connection(agent_name: str) -> A2AConnection | None

Get the A2A connection for a given remote agent name, if it uses A2A protocol.

Arguments:

• agent_name str - The name of the remote agent.

Returns:

A2AConnection | None: The A2A connection object if the agent is found and uses A2A protocol, otherwise None.

list_tools

async def list_tools(mcp_server_names: List[str]) -> List[BaseTool]

Retrieve tools from the specified MCP servers.

Arguments:

• mcp_server_names List[str] - List of MCP server names to retrieve tools from.

Returns:

• List[BaseTool] - List of tools available from the specified MCP servers.

Raises:

• ConnectionError - If a required MCP server cannot be connected to.

list_agent_cards

async def list_agent_cards(agent_names: List[str]) -> Dict[str, AgentCard]

Retrieve agent cards from the specified remote agents. If a non-required remote agent cannot be reached, it is not included in the result.

Arguments:

• agent_names List[str] - List of remote agent names to retrieve agent cards from.

Returns:

Dict[str, AgentCard]: Dictionary of Agent Cards available from the specified remote agents.

Raises:

• ConnectionError - If a required remote agent cannot be connected to.

bat.agent.graph

AgentGraph Objects

class AgentGraph(ABC)

Abstract base class for agent graphs.

Extend this class to implement the specific behavior of an agent: add the nodes and edges of the graph in setup. The graph is instantiated and compiled by AgentApplication, so subclasses do not override __init__.

Example

from bat.agent import AgentConfig, AgentGraph, AgentState
from langgraph.graph import END, START

class MyAgentState(AgentState):
    # Your state here
    # ...
    pass

class MyAgentGraph(AgentGraph):
    def setup(self, config: AgentConfig) -> None:
        # self.graph_builder is a StateGraph of the agent state type
        self.graph_builder.add_node("answer", self.answer)
        self.graph_builder.add_edge(START, "answer")
        self.graph_builder.add_edge("answer", END)

    # Your nodes logic here
    def answer(self, state: MyAgentState) -> MyAgentState:
        ...

__init__

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

Initialize the AgentGraph: create a state graph of StateType, let setup add its nodes and edges, then compile it.

The graph is compiled with an in-memory checkpointer when config.checkpoints is set. Called by AgentApplication.

Arguments:

• config AgentConfig - The agent configuration, passed to setup.
• StateType Type[AgentState] - The AgentState schema of the graph.

graph_builder

@property
def graph_builder() -> StateGraph

Get the state graph builder.

Returns:

• StateGraph - The state graph builder.

compiled_graph

@property
def compiled_graph() -> CompiledStateGraph

Get the compiled state graph.

Returns:

• CompiledStateGraph - The compiled state graph of the agent.

setup

@abstractmethod
def setup(config: AgentConfig) -> None

Set up the agent graph with the provided configuration. Subclasses must implement this method.

Arguments:

• config AgentConfig - The agent configuration.

astream

async def astream(query: str,
                  config: RunnableConfig) -> AsyncIterable[AgentTaskResult]

Asynchronously stream results from the agent graph based on the query and configuration.

While the graph runs, WORKING results are yielded as progress, without consecutive duplicates. When it stops, exactly one final result is yielded:

• INPUT_REQUIRED if the graph is paused on an interrupt;
• otherwise to_task_result() of the graph's final state;
• FAILED if the graph raised or finished without a final result.

Other results of intermediate states are ignored: the answer always comes from the final state. The executor publishes every result yielded here, so an override must follow these rules.

This method performs the following steps:

1. Looks for a checkpoint associated with the provided configuration.
2. If no checkpoint is found, creates a new agent state from the query, using the from_query method of the StateType.
3. If a checkpoint is found, restores the state from the checkpoint and updates it with the query using the update_after_checkpoint_restore method. A checkpoint that is not a valid StateType is dropped: the state starts over from the query.
4. Prepares the input for the graph execution, wrapping the state in a Command if the is_waiting_for_human_input method of the state returns True.
5. Executes the graph with the astream method, including nested graphs, so that prebuilt workflows can report progress.
6. Converts each streamed state with to_task_result, and yields it if it is new WORKING progress. A state of a nested graph that is not a StateType is skipped.
7. Once the graph stops, yields the final result.

This method prints debug logs in the format [<thread_id>]: <message>.

Arguments:

• query str - The query to process.
• config RunnableConfig - Configuration for the runnable.

Returns:

• AsyncIterable[AgentTaskResult] - An asynchronous iterable of agent task results.

draw_mermaid

def draw_mermaid(file_path: Optional[str] = None) -> None

Draw the agent graph in Mermaid format. If a file path is provided, save the diagram to the file, otherwise print it to the console.

Arguments:

• file_path Optional[str] - The path to the file where the Mermaid diagram should be saved.

build_message

@staticmethod
def build_message(config: RunnableConfig, text: str) -> Message

Build a user A2A text message to send to a remote agent.

Fits the build_message callback of CallAgentNode. The message takes its context_id from the thread_id of the configuration (default if unset) and its task_id from the task_id, if any.

Arguments:

• config RunnableConfig - The configuration of the running graph.
• text str - The text of the message.

Returns:

• Message - The A2A message.