a2a_platform#

class baf.platforms.a2a.a2a_platform.A2APlatform(agent, version='1.0', capabilities=None, id=None, endpoints=None, descriptions=None, skills=None, examples=None, methods=None, provider='BESSER-Agentic-Framework')[source]#

Bases: Platform

A2APlatform implements the Agent-to-Agent (A2A) communication platform for agent interactions. This platform enables agents to communicate with each other using a RESTful API and Server-Sent Events (SSE) over HTTP. It stores agent metadata (e.g., capabilities and examples), registers methods, maintains task creation, execution and status tracking, supports tracked and untracked agent-to-agent calls, provides synchronous and asynchronous operations, and supports agent orchestration for coordinating multiple agents.

Parameters:

agent (Agent) – The agent instance this platform belongs to
version (str, optional) – Version of the agent. Defaults to ‘1.0’
capabilities (list[str], optional) – List of agent capabilities. Defaults to []
id (str, optional) – Unique identifier for the agent. Defaults to UUID4
endpoints (list[str], optional) – List of API endpoints. Defaults to standard endpoints at localhost:8000
descriptions (list[str], optional) – List of agent descriptions. Defaults to []
skills (list[str], optional) – List of agent skills. Defaults to []
examples (Union[list[dict], list[str]], optional) – Usage examples. Defaults to []
methods (list[dict], optional) – List of available methods. Defaults to []
provider (str, optional) – Provider name. Defaults to “BESSER-Agentic-Framework”

_agent#

The agent this platform belongs to

Type:: Agent

agent_card#

Contains metadata about the agent’s capabilities, interface, connection and so on

Type:: AgentCard

router#

Handles routing of messages between agents and agent to endpoints via RPC

Type:: A2ARouter

tasks#

Storage for registering tasks, managing and monitoring their states

Type:: dict

_port#

Port number on which this platform will be hosted

Type:: int

_app#

Web application to run this platform

Type:: web.Application

_abc_impl = <_abc._abc_data object>#

_send(session, payload)[source]#: str: log _send() method not implemented in class

add_capabilities(capability)[source]#: Helper function to add agent’s capabilities to the agent_card

add_descriptions(descriptions)[source]#: Helper function to add agent’s description to the agent_card

add_examples(examples)[source]#: Helper function to add agent execution examples to the agent_card

add_methods(methods)[source]#: Helper function to add agent’s methods manually to the agent_card.

add_skills(skills)[source]#: Helper function to add agent’s skills to the agent_card

async create_and_execute_task(method, params)[source]#: This is an internal method. It creates a task and runs it in the background (asynchronous).

create_task(method, params)[source]#

async execute_task(task_id)[source]#

get_agent_card()[source]#: Returns the agent card in JSON format.

get_status(task_id)[source]#

initialize()[source]#: Initializes the platform

list_tasks()[source]#

populate_methods_from_router()[source]#: Automatically fetch agent’s registered methods from router and add it to the agent_card.

register_orchestration_as_task(name, coroutine_func, registry)[source]#: Wrap an async orchestration function as a tracked task. Tracking can be done in the orchestration agent_id/tasks endpoint. Backward compatible with coroutine_func.

register_orchestration_task_on_resp_agent(name, func, registry)[source]#: This function is only for async execution of multiple agents, does not register the execution as a task in Orchestrator’s task endpoint. Will register tasks on the respective agent’s router. So tasks can be tracked only in their respective agent_id/tasks.

reply(session, message)[source]#: str: log reply() method not implemented in class

async rpc_call_agent(target_agent_id, method, params, registry)[source]#: Calls another agent as a subtask, waits for it to complete, and returns its task info, results and so on. Orchestration task cannot track subtask statuses. They can be tracked in the respective agent_id/tasks endpoint.

async rpc_call_agent_tracked(target_agent_id, method, params, registry, parent_task=None)[source]#: Calls another agent as a subtask, waits for it to complete, and returns its task info, results and so on. Ensures the orchestration task can track subtask statuses at orchestrator agent_id/tasks endpoint.

async rpc_create_task(method, params)[source]#: This is an internal method. It creates an asynchronous task and set the status to PENDING execution or RUNNING depending on the tasks queued in the server. Once the execution is done, results will be available here.

start()[source]#: Starts the platform

stop()[source]#: Stops the platform