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 remote A2A or MCP agent connections.

Attributes:

• name str - The name of the remote agent.

• url str - The URL of the remote agent.

• protocol InteragentCommunicationProtocol - The communication protocol used by the remote agent ('A2A' or 'MCP').

• 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.

AgentConfig Objects

class AgentConfig(BaseModel)

Agent Configuration, including MCP servers and remote agents.

Attributes

mcp_servers (List[MCPServerConfig]): List of MCP server
    configurations.
remote_agents (List[RemoteAgentConfig]): List of remote agent
    configurations.

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.

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_mcp_agent_connection

def get_mcp_agent_connection(agent_name: str) -> MCPConnection | None

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

Arguments:

• agent_name str - The name of the remote agent.

Returns:

MCPConnection | None: The MCP connection object if the agent is found and uses MCP protocol, 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.application

AgentApplication Objects

class AgentApplication()

Agent Application based on Starlette. This class sets up an agent application that can handle A2A and MCP protocols.

Supported Environment Variables: - URL (required): The base URL where the agent will be hosted. - PORT: The port for the A2A application. Defaults to 9900. - MCP_PORT: The port for the MCP application. Defaults to 9800. - CONFIG: Path to a configuration file for the agent. Defaults to "config.yaml". - AGENT_CARD_PATH: Path to the agent card. Defaults to "./agent.json" - AGENT_CARD_DISPLAY: Whether to display the AgentCard when the agent starts. Defaults to True.

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])

Initialize the AgentApplication with the given agent card path and agent graph.

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.

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 the URL environment variable is not set.
• 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(expose_mcp: bool = False) -> None

Run the agent application.

Arguments:

• expose_mcp bool, optional - Whether to expose the MCP protocol. Defaults to False. This parameter isn't fully supported yet and may lead to unexpected behavior when set to True.

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.

Example

from bat.agent import AgentGraph, AgentState
from langgraph.runnables import RunnableConfig
from langgraph.graph import StateGraph

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

class MyAgentGraph(AgentGraph):
    def __init__(self):
        # Define the agent graph using langgraph.graph.StateGraph class
        graph_builder = StateGraph(MyAgentState)
        # Add nodes and edges to the graph as needed ...
        super().__init__(
            graph_builder=graph_builder,
            use_checkpoint=True,
            logger_name="my_agent"
        )
        self._log("Graph initialized", "info")

    # Your nodes logic here
    # ...

__init__

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

Initialize the AgentGraph with a state graph and optional checkpointing and logger. Compile the state graph and set up the logger if the logger_name is provided.

Arguments:

• graph_builder StateGraph - The state graph builder.
• use_checkpoint bool - Whether to use checkpointing. Defaults to False.
• logger_name Optional[str] - The name of the logger to use. Defaults to None.

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. This method performes 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.
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, passing the input and configuration.
6. For each item in the stream:

• If it is an interrupt, yields an AgentTaskResult with the status input-required. This enables human-in-the-loop interactions.
• Otherwise, validates the item as an StateType and converts it to an AgentTaskResult using the to_task_result method of the state. Then it yields the 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.

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:

• item 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
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 restoring from a saved checkpoint. Allows the state to synchronize with new execution parameters before resuming the graph.

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.

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