A2A (Agent-to-Agent)¶

A2A (Agent-to-Agent) Protocol Implementation for AI-Parrot.

Exposes AI-Parrot agents as A2A-compliant microservices, enabling inter-agent communication across network boundaries.

Components

Server Layer: - A2AServer: Wrap an agent as an A2A HTTP service - A2AEnabledMixin: Mixin to add A2A server capabilities

Client Layer: - A2AClient: Connect to remote A2A agents - A2AClientMixin: Mixin to add A2A client capabilities - A2AAgentConnection: Connection state for remote agents - A2ARemoteAgentTool: Use remote agent as a tool - A2ARemoteSkillTool: Use remote skill as a tool

Discovery Layer: - A2AMeshDiscovery: Centralized agent discovery service - A2AEndpoint: Configuration for an A2A endpoint - HealthCheckStrategy: How to check agent health - AgentStatus: Agent health status

Routing Layer: - A2AProxyRouter: Rule-based request routing - RoutingStrategy: How to match requests to agents - LoadBalanceStrategy: How to balance across agents - RoutingRule: Individual routing rule definition

Orchestration Layer: - A2AOrchestrator: Hybrid routing with LLM fallback - OrchestrationMode: Execution modes (rules, llm, hybrid, etc.) - OrchestrationResult: Result from orchestration - OrchestrationPlan: LLM decision plan

Security Layer (server-only — requires ai-parrot-server): - AuthScheme: Supported authentication schemes - CallerIdentity: Authenticated agent identity - SecurityPolicy: Access control policies - CredentialProvider: Abstract credential storage - InMemoryCredentialProvider: Dev/test credential store - RedisCredentialProvider: Production credential store - JWTAuthenticator: JWT token authentication - MTLSAuthenticator: Mutual TLS authentication - A2ASecurityMiddleware: Security middleware for servers - SecureA2AClient: Client with automatic authentication

Data Models: - AgentCard: Agent metadata and capabilities - AgentSkill: Skill/capability exposed by agent - AgentCapabilities: Agent feature flags - Task: A2A task with status and artifacts - Message: A2A message format - Artifact: Output produced by agent

A2AClient ¶

A2AClient(base_url: str, *, timeout: float = 60.0, headers: Optional[Dict[str, str]] = None, auth_token: Optional[str] = None, api_key: Optional[str] = None)

Client for communicating with remote A2A agents.

Example

async with A2AClient("https://remote-agent:8080") as client: # Discover agent card = await client.discover() print(f"Connected to: {card.name}")

# Send message
task = await client.send_message("Hello!")
print(task.artifacts[0].parts[0].text)

# Stream response
async for chunk in client.stream_message("Explain quantum computing"):
    print(chunk, end="", flush=True)

Initialize A2A client.

PARAMETER	DESCRIPTION
`base_url`	Base URL of the A2A agent (e.g., "https://agent.example.com") TYPE: `str`
`timeout`	Request timeout in seconds TYPE: `float` DEFAULT: `60.0`
`headers`	Additional headers to send with requests TYPE: `Optional[Dict[str, str]]` DEFAULT: `None`
`auth_token`	Bearer token for authentication TYPE: `Optional[str]` DEFAULT: `None`
`api_key`	API key for authentication (sent as X-API-Key header) TYPE: `Optional[str]` DEFAULT: `None`

Source code in packages/ai-parrot/src/parrot/a2a/client.py

def __init__(
    self,
    base_url: str,
    *,
    timeout: float = 60.0,
    headers: Optional[Dict[str, str]] = None,
    auth_token: Optional[str] = None,
    api_key: Optional[str] = None,
):
    """
    Initialize A2A client.

    Args:
        base_url: Base URL of the A2A agent (e.g., "https://agent.example.com")
        timeout: Request timeout in seconds
        headers: Additional headers to send with requests
        auth_token: Bearer token for authentication
        api_key: API key for authentication (sent as X-API-Key header)
    """
    self.base_url = base_url.rstrip("/")
    self.timeout = aiohttp.ClientTimeout(total=timeout)
    self._session: Optional[aiohttp.ClientSession] = None
    self._agent_card: Optional[AgentCard] = None
    self._owns_session = False
    # Protocol version spoken by the remote server, detected during
    # discover() from the AgentCard shape ("1.0" or "0.3").
    self._server_version: Optional[str] = None

    # Build headers. The client speaks A2A v1.0 by default and advertises
    # it via the `A2A-Version` header (FEAT-272 / TASK-1717).
    self.headers = {"Content-Type": "application/json", "A2A-Version": "1.0"}
    if headers:
        self.headers.update(headers)
    if auth_token:
        self.headers["Authorization"] = f"Bearer {auth_token}"
    if api_key:
        self.headers["X-API-Key"] = api_key
    # Public alias for tests/introspection.
    self._default_headers = self.headers

    self.logger = logging.getLogger(f"A2AClient.{base_url}")

agent_card `property` ¶

agent_card: Optional[AgentCard]

Get the cached agent card.

connect `async` ¶

connect(session: Optional[ClientSession] = None) -> None

Establish connection and discover remote agent.

Source code in packages/ai-parrot/src/parrot/a2a/client.py

async def connect(self, session: Optional[aiohttp.ClientSession] = None) -> None:
    """Establish connection and discover remote agent."""
    if session:
        self._session = session
        self._owns_session = False
    else:
        self._session = aiohttp.ClientSession(
            timeout=self.timeout,
            headers=self.headers
        )
        self._owns_session = True

    # Discover agent card
    self._agent_card = await self.discover()
    self.logger.info(f"Connected to A2A agent: {self._agent_card.name}")

disconnect `async` ¶

disconnect() -> None

Close the connection.

Source code in packages/ai-parrot/src/parrot/a2a/client.py

async def disconnect(self) -> None:
    """Close the connection."""
    if self._session and self._owns_session:
        await self._session.close()
        self._session = None
    self._agent_card = None

discover `async` ¶

discover() -> AgentCard

Fetch the remote agent's card.

Tries the A2A v1.0 discovery URI (/.well-known/agent-card.json) first and falls back to the v0.3 URI (/.well-known/agent.json). The server's protocol version is detected from the card shape (presence of supportedInterfaces => v1.0) and cached on self._server_version.

Source code in packages/ai-parrot/src/parrot/a2a/client.py

async def discover(self) -> AgentCard:
    """Fetch the remote agent's card.

    Tries the A2A v1.0 discovery URI (``/.well-known/agent-card.json``)
    first and falls back to the v0.3 URI (``/.well-known/agent.json``).
    The server's protocol version is detected from the card shape
    (presence of ``supportedInterfaces`` => v1.0) and cached on
    ``self._server_version``.
    """
    data: Optional[Dict[str, Any]] = None

    # v1.0 discovery URI first.
    try:
        async with self._session.get(
            f"{self.base_url}/.well-known/agent-card.json"
        ) as resp:
            if resp.status == 200:
                data = await resp.json()
    except aiohttp.ClientError:
        data = None

    # Fall back to the v0.3 discovery URI.
    if data is None:
        async with self._session.get(
            f"{self.base_url}/.well-known/agent.json"
        ) as resp:
            resp.raise_for_status()
            data = await resp.json()

    self._server_version = "1.0" if "supportedInterfaces" in data else "0.3"

    # AgentCard.from_dict auto-detects the v1.0/v0.3 shape.
    card = AgentCard.from_dict(data)
    # Guarantee a usable URL even when the card omits one.
    if not card.supported_interfaces:
        from .models import AgentInterface
        card.supported_interfaces = [
            AgentInterface(url=self.base_url, protocol_binding="JSONRPC")
        ]
    return card

get_skills ¶

get_skills() -> List[AgentSkill]

Get available skills from the remote agent.

Source code in packages/ai-parrot/src/parrot/a2a/client.py

def get_skills(self) -> List[AgentSkill]:
    """Get available skills from the remote agent."""
    if not self._agent_card:
        return []
    return self._agent_card.skills

get_skill ¶

get_skill(skill_id: str) -> Optional[AgentSkill]

Get a specific skill by ID.

Source code in packages/ai-parrot/src/parrot/a2a/client.py

def get_skill(self, skill_id: str) -> Optional[AgentSkill]:
    """Get a specific skill by ID."""
    for skill in self.get_skills():
        if skill.id == skill_id:
            return skill
    return None

send_message `async` ¶

send_message(content: str, *, context_id: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None) -> Task

Send a message to the remote agent and wait for response.

PARAMETER	DESCRIPTION
`content`	The message content TYPE: `str`
`context_id`	Optional context ID for multi-turn conversations TYPE: `Optional[str]` DEFAULT: `None`
`metadata`	Optional metadata to include TYPE: `Optional[Dict[str, Any]]` DEFAULT: `None`

RETURNS	DESCRIPTION
`Task`	Task with the response

Source code in packages/ai-parrot/src/parrot/a2a/client.py

async def send_message(
    self,
    content: str,
    *,
    context_id: Optional[str] = None,
    metadata: Optional[Dict[str, Any]] = None,
) -> Task:
    """
    Send a message to the remote agent and wait for response.

    Args:
        content: The message content
        context_id: Optional context ID for multi-turn conversations
        metadata: Optional metadata to include

    Returns:
        Task with the response
    """
    url = f"{self.base_url}/a2a/message{self._verb_sep}send"

    message = Message.user(content, context_id=context_id, metadata=metadata)

    payload = {
        "message": message.to_dict()
    }

    async with self._session.post(url, json=payload) as resp:
        resp.raise_for_status()
        data = await resp.json()

    return self._parse_task(data)

stream_message `async` ¶

stream_message(content: str, *, context_id: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None) -> AsyncIterator[str]

Send a message and stream the response.

PARAMETER	DESCRIPTION
`content`	The message content TYPE: `str`
`context_id`	Optional context ID TYPE: `Optional[str]` DEFAULT: `None`
`metadata`	Optional metadata TYPE: `Optional[Dict[str, Any]]` DEFAULT: `None`

YIELDS	DESCRIPTION
`AsyncIterator[str]`	Text chunks as they arrive

Source code in packages/ai-parrot/src/parrot/a2a/client.py

async def stream_message(
    self,
    content: str,
    *,
    context_id: Optional[str] = None,
    metadata: Optional[Dict[str, Any]] = None,
) -> AsyncIterator[str]:
    """
    Send a message and stream the response.

    Args:
        content: The message content
        context_id: Optional context ID
        metadata: Optional metadata

    Yields:
        Text chunks as they arrive
    """
    url = f"{self.base_url}/a2a/message{self._verb_sep}stream"

    message = Message.user(content, context_id=context_id, metadata=metadata)

    payload = {
        "message": message.to_dict()
    }

    async with self._session.post(
        url,
        json=payload,
        headers={"Accept": "text/event-stream"}
    ) as resp:
        resp.raise_for_status()

        async for line in resp.content:
            line = line.decode("utf-8").strip()

            if not line or not line.startswith("data:"):
                continue

            try:
                data = json.loads(line[5:].strip())

                # Extract text from artifact updates
                if "artifactUpdate" in data:
                    artifact = data["artifactUpdate"].get("artifact", {})
                    parts = artifact.get("parts", [])
                    for part in parts:
                        if "text" in part:
                            yield part["text"]

                # Check for completion/failure
                if "statusUpdate" in data:
                    status = data["statusUpdate"]
                    if status.get("final"):
                        state = status.get("status", {}).get("state")
                        # Accept both v0.3 "failed" and v1.0 "TASK_STATE_FAILED".
                        if state and parse_task_state(state) == TaskState.FAILED:
                            error_msg = status.get("status", {}).get("message", {})
                            error_text = error_msg.get("parts", [{}])[0].get("text", "Unknown error")
                            raise RuntimeError(f"Remote agent failed: {error_text}")
                        break

            except json.JSONDecodeError:
                continue

invoke_skill `async` ¶

invoke_skill(skill_id: str, params: Optional[Dict[str, Any]] = None, *, context_id: Optional[str] = None) -> Any

Invoke a specific skill on the remote agent.

PARAMETER	DESCRIPTION
`skill_id`	The skill ID to invoke TYPE: `str`
`params`	Parameters to pass to the skill TYPE: `Optional[Dict[str, Any]]` DEFAULT: `None`
`context_id`	Optional context ID TYPE: `Optional[str]` DEFAULT: `None`

RETURNS	DESCRIPTION
`Any`	The skill result (extracted from artifacts)

Source code in packages/ai-parrot/src/parrot/a2a/client.py

async def invoke_skill(
    self,
    skill_id: str,
    params: Optional[Dict[str, Any]] = None,
    *,
    context_id: Optional[str] = None,
) -> Any:
    """
    Invoke a specific skill on the remote agent.

    Args:
        skill_id: The skill ID to invoke
        params: Parameters to pass to the skill
        context_id: Optional context ID

    Returns:
        The skill result (extracted from artifacts)
    """
    message = Message.user(
        {"skill": skill_id, "params": params or {}},
        context_id=context_id
    )

    url = f"{self.base_url}/a2a/message{self._verb_sep}send"
    payload = {"message": message.to_dict()}

    async with self._session.post(url, json=payload) as resp:
        resp.raise_for_status()
        data = await resp.json()

    task = self._parse_task(data)

    if task.status.state == TaskState.FAILED:
        error_msg = task.status.message.get_text() if task.status.message else "Unknown error"
        raise RuntimeError(f"Skill invocation failed: {error_msg}")

    # Extract result from artifacts
    if task.artifacts:
        artifact = task.artifacts[0]
        if artifact.parts:
            part = artifact.parts[0]
            if part.data:
                return part.data
            return part.text

    return None

get_task `async` ¶

get_task(task_id: str) -> Task

Get a task by ID.

Source code in packages/ai-parrot/src/parrot/a2a/client.py

async def get_task(self, task_id: str) -> Task:
    """Get a task by ID."""
    url = f"{self.base_url}/a2a/tasks/{task_id}"

    async with self._session.get(url) as resp:
        resp.raise_for_status()
        data = await resp.json()

    return self._parse_task(data)

list_tasks `async` ¶

list_tasks(context_id: Optional[str] = None, status: Optional[str] = None, page_size: int = 50) -> List[Task]

List tasks with optional filtering.

Source code in packages/ai-parrot/src/parrot/a2a/client.py

async def list_tasks(
    self,
    context_id: Optional[str] = None,
    status: Optional[str] = None,
    page_size: int = 50,
) -> List[Task]:
    """List tasks with optional filtering."""
    url = f"{self.base_url}/a2a/tasks"
    params = {"pageSize": page_size}
    if context_id:
        params["contextId"] = context_id
    if status:
        params["status"] = status

    async with self._session.get(url, params=params) as resp:
        resp.raise_for_status()
        data = await resp.json()

    return [self._parse_task(t) for t in data.get("tasks", [])]

cancel_task `async` ¶

cancel_task(task_id: str) -> Task

Cancel a running task.

Source code in packages/ai-parrot/src/parrot/a2a/client.py

async def cancel_task(self, task_id: str) -> Task:
    """Cancel a running task."""
    url = f"{self.base_url}/a2a/tasks/{task_id}{self._verb_sep}cancel"

    async with self._session.post(url) as resp:
        resp.raise_for_status()
        data = await resp.json()

    return self._parse_task(data)

create_push_config `async` ¶

create_push_config(task_id: str, url: str, *, config_id: str = '', authentication: Optional[Dict[str, Any]] = None, metadata: Optional[Dict[str, Any]] = None) -> TaskPushNotificationConfig

Register a push-notification webhook config for a task.

Source code in packages/ai-parrot/src/parrot/a2a/client.py

async def create_push_config(
    self,
    task_id: str,
    url: str,
    *,
    config_id: str = "",
    authentication: Optional[Dict[str, Any]] = None,
    metadata: Optional[Dict[str, Any]] = None,
) -> TaskPushNotificationConfig:
    """Register a push-notification webhook config for a task."""
    endpoint = f"{self.base_url}/a2a/tasks/{task_id}/pushNotificationConfigs"
    payload: Dict[str, Any] = {"id": config_id, "taskId": task_id, "url": url}
    if authentication is not None:
        payload["authentication"] = authentication
    if metadata is not None:
        payload["metadata"] = metadata

    async with self._session.post(endpoint, json=payload) as resp:
        resp.raise_for_status()
        data = await resp.json()
    return TaskPushNotificationConfig.from_dict(data)

get_push_config `async` ¶

get_push_config(task_id: str, config_id: str) -> TaskPushNotificationConfig

Fetch a single push-notification config.

Source code in packages/ai-parrot/src/parrot/a2a/client.py

async def get_push_config(
    self, task_id: str, config_id: str
) -> TaskPushNotificationConfig:
    """Fetch a single push-notification config."""
    endpoint = (
        f"{self.base_url}/a2a/tasks/{task_id}/pushNotificationConfigs/{config_id}"
    )
    async with self._session.get(endpoint) as resp:
        resp.raise_for_status()
        data = await resp.json()
    return TaskPushNotificationConfig.from_dict(data)

list_push_configs `async` ¶

list_push_configs(task_id: str) -> List[TaskPushNotificationConfig]

List all push-notification configs registered for a task.

Source code in packages/ai-parrot/src/parrot/a2a/client.py

async def list_push_configs(
    self, task_id: str
) -> List[TaskPushNotificationConfig]:
    """List all push-notification configs registered for a task."""
    endpoint = f"{self.base_url}/a2a/tasks/{task_id}/pushNotificationConfigs"
    async with self._session.get(endpoint) as resp:
        resp.raise_for_status()
        data = await resp.json()
    return [
        TaskPushNotificationConfig.from_dict(c) for c in data.get("configs", [])
    ]

delete_push_config `async` ¶

delete_push_config(task_id: str, config_id: str) -> bool

Delete a push-notification config; returns True on success.

Source code in packages/ai-parrot/src/parrot/a2a/client.py

async def delete_push_config(self, task_id: str, config_id: str) -> bool:
    """Delete a push-notification config; returns True on success."""
    endpoint = (
        f"{self.base_url}/a2a/tasks/{task_id}/pushNotificationConfigs/{config_id}"
    )
    async with self._session.delete(endpoint) as resp:
        resp.raise_for_status()
        data = await resp.json()
    return bool(data.get("deleted", False))

rpc_call `async` ¶

rpc_call(method: str, params: Optional[Dict[str, Any]] = None) -> Any

Make a JSON-RPC call to the remote agent.

Source code in packages/ai-parrot/src/parrot/a2a/client.py

async def rpc_call(
    self,
    method: str,
    params: Optional[Dict[str, Any]] = None,
) -> Any:
    """Make a JSON-RPC call to the remote agent."""
    url = f"{self.base_url}/a2a/rpc"

    payload = {
        "jsonrpc": "2.0",
        "id": str(id(self)),
        "method": method,
        "params": params or {}
    }

    async with self._session.post(url, json=payload) as resp:
        resp.raise_for_status()
        data = await resp.json()

    if "error" in data:
        raise RuntimeError(f"RPC error: {data['error']}")

    return data.get("result")

A2AAgentConnection `dataclass` ¶

A2AAgentConnection(url: str, card: AgentCard, client: 'A2AClient', name: str = '')

Represents a connection to a remote A2A agent.

A2ARemoteAgentTool ¶

A2ARemoteAgentTool(client: A2AClient, *, tool_name: Optional[str] = None, tool_description: Optional[str] = None, use_streaming: bool = False, **kwargs)

Bases: AbstractTool

Wraps a remote A2A agent as a tool that can be used by local agents.

This creates a tool that, when invoked, sends the query to the remote agent. Properly inherits from AbstractTool for ToolManager compatibility.

Initialize A2A Remote Agent Tool.

PARAMETER	DESCRIPTION
`client`	Connected A2AClient instance TYPE: `A2AClient`
`tool_name`	Custom name for the tool (defaults to ask_) TYPE: `Optional[str]` DEFAULT: `None`
`tool_description`	Custom description (defaults to agent card description) TYPE: `Optional[str]` DEFAULT: `None`
`use_streaming`	Whether to use streaming for responses TYPE: `bool` DEFAULT: `False`

Source code in packages/ai-parrot/src/parrot/a2a/client.py

def __init__(
    self,
    client: A2AClient,
    *,
    tool_name: Optional[str] = None,
    tool_description: Optional[str] = None,
    use_streaming: bool = False,
    **kwargs
):
    """
    Initialize A2A Remote Agent Tool.

    Args:
        client: Connected A2AClient instance
        tool_name: Custom name for the tool (defaults to ask_<agent_name>)
        tool_description: Custom description (defaults to agent card description)
        use_streaming: Whether to use streaming for responses
    """
    self.client = client
    self.use_streaming = use_streaming
    self.tags = ["a2a", "remote-agent"]

    card = client.agent_card
    name = tool_name or f"ask_{card.name.lower().replace(' ', '_')}"
    description = tool_description or (
        f"Ask the remote agent '{card.name}': {card.description}"
    )

    # Store for cloning
    self._tool_name_override = tool_name
    self._tool_description_override = tool_description

    super().__init__(name=name, description=description, **kwargs)

clone ¶

clone() -> 'A2ARemoteAgentTool'

Clone this tool (shares the client reference).

Source code in packages/ai-parrot/src/parrot/a2a/client.py

def clone(self) -> "A2ARemoteAgentTool":
    """Clone this tool (shares the client reference)."""
    clone_kwargs = self._get_clone_kwargs()
    # Remove standard AbstractTool params that we handle differently
    clone_kwargs.pop('name', None)
    clone_kwargs.pop('description', None)
    return A2ARemoteAgentTool(client=self.client, **clone_kwargs)

A2ARemoteSkillTool ¶

A2ARemoteSkillTool(client: A2AClient, skill: AgentSkill, **kwargs)

Bases: AbstractTool

Wraps a specific skill from a remote A2A agent as a tool.

Properly inherits from AbstractTool for ToolManager compatibility. Dynamically generates input schema from the skill's input_schema.

Initialize A2A Remote Skill Tool.

PARAMETER	DESCRIPTION
`client`	Connected A2AClient instance TYPE: `A2AClient`
`skill`	The AgentSkill to wrap as a tool TYPE: `AgentSkill`

Source code in packages/ai-parrot/src/parrot/a2a/client.py

def __init__(
    self,
    client: A2AClient,
    skill: AgentSkill,
    **kwargs
):
    """
    Initialize A2A Remote Skill Tool.

    Args:
        client: Connected A2AClient instance
        skill: The AgentSkill to wrap as a tool
    """
    self.client = client
    self.skill = skill
    self.tags = ["a2a", "remote-skill"] + skill.tags

    name = f"remote_{skill.id}"
    description = skill.description or f"Invoke remote skill: {skill.name}"

    # Dynamically create the args schema from skill.input_schema
    self.args_schema = _create_skill_input_model(skill)

    super().__init__(name=name, description=description, **kwargs)

clone ¶

clone() -> 'A2ARemoteSkillTool'

Clone this tool (shares the client and skill references).

Source code in packages/ai-parrot/src/parrot/a2a/client.py

def clone(self) -> "A2ARemoteSkillTool":
    """Clone this tool (shares the client and skill references)."""
    clone_kwargs = self._get_clone_kwargs()
    return A2ARemoteSkillTool(client=self.client, skill=self.skill, **clone_kwargs)

A2AClientMixin ¶

A2AClientMixin(*args, **kwargs)

Mixin to add A2A client capabilities to any AbstractBot.

This allows an agent to communicate with remote A2A agents, either directly or by registering them as tools.

Features

Direct connection to remote A2A agents
Integration with A2AMeshDiscovery for centralized discovery
Integration with A2AProxyRouter for rule-based routing
Integration with A2AOrchestrator for hybrid orchestration
Automatic tool registration for remote agents/skills

Example

class MyAgent(A2AClientMixin, BasicAgent): pass

agent = MyAgent(name="Orchestrator", llm="openai:gpt-4") await agent.configure()

Option 1: Connect to remote agents directly¶

await agent.add_a2a_agent("https://data-agent:8080") await agent.add_a2a_agent("https://search-agent:8081")

Now the agent can use remote agents as tools¶

response = await agent.ask("Search for X and analyze the data")

Or call remote agents directly¶

result = await agent.ask_remote_agent("data-agent", "What's the total revenue?")

Option 2: Use mesh discovery¶

mesh = A2AMeshDiscovery.from_config("agents.yaml") await mesh.start() agent.set_mesh(mesh) await agent.discover_from_mesh(skill="data_analysis")

Option 3: Use router for rule-based routing¶

router = A2AProxyRouter(mesh) router.route_by_skill("analysis", "AnalystBot") agent.set_router(router) result = await agent.route_to_agent("Analyze this data")

Option 4: Use orchestrator for hybrid routing¶

orchestrator = A2AOrchestrator(mesh) orchestrator.set_fallback_llm(llm_client) agent.set_orchestrator(orchestrator) result = await agent.orchestrate("Complex multi-agent task")

Source code in packages/ai-parrot/src/parrot/a2a/mixin.py

def __init__(self, *args, **kwargs):
    super().__init__(*args, **kwargs)
    self._a2a_clients: Dict[str, A2AAgentConnection] = {}
    self._a2a_mesh: Optional["A2AMeshDiscovery"] = None
    self._a2a_router: Optional["A2AProxyRouter"] = None
    self._a2a_orchestrator: Optional["A2AOrchestrator"] = None
    self._a2a_matrix_transport: Optional[Any] = None  # MatrixA2ATransport
    self._a2a_logger = logging.getLogger(
        f"A2AClient.{getattr(self, 'name', 'Agent')}"
    )

set_matrix_transport ¶

set_matrix_transport(transport: Any) -> None

Set a Matrix-based A2A transport.

When configured, the agent can use Matrix rooms for A2A communication instead of (or in addition to) HTTP.

PARAMETER	DESCRIPTION
`transport`	MatrixA2ATransport instance. TYPE: `Any`

Source code in packages/ai-parrot/src/parrot/a2a/mixin.py

def set_matrix_transport(self, transport: Any) -> None:
    """Set a Matrix-based A2A transport.

    When configured, the agent can use Matrix rooms for A2A
    communication instead of (or in addition to) HTTP.

    Args:
        transport: MatrixA2ATransport instance.
    """
    self._a2a_matrix_transport = transport
    self._a2a_logger.info("Connected to Matrix A2A transport")

get_matrix_transport ¶

get_matrix_transport() -> Optional[Any]

Get the configured Matrix A2A transport.

Source code in packages/ai-parrot/src/parrot/a2a/mixin.py

def get_matrix_transport(self) -> Optional[Any]:
    """Get the configured Matrix A2A transport."""
    return self._a2a_matrix_transport

set_mesh ¶

set_mesh(mesh: 'A2AMeshDiscovery') -> None

Connect this agent to an A2A mesh discovery service.

The mesh provides centralized discovery and health checking of remote A2A agents.

PARAMETER	DESCRIPTION
`mesh`	A2AMeshDiscovery instance TYPE: `'A2AMeshDiscovery'`

Example

mesh = A2AMeshDiscovery.from_config("agents.yaml") await mesh.start() agent.set_mesh(mesh)

Source code in packages/ai-parrot/src/parrot/a2a/mixin.py

def set_mesh(self, mesh: "A2AMeshDiscovery") -> None:
    """
    Connect this agent to an A2A mesh discovery service.

    The mesh provides centralized discovery and health checking
    of remote A2A agents.

    Args:
        mesh: A2AMeshDiscovery instance

    Example:
        mesh = A2AMeshDiscovery.from_config("agents.yaml")
        await mesh.start()
        agent.set_mesh(mesh)
    """
    self._a2a_mesh = mesh
    self._a2a_logger.info("Connected to A2A mesh discovery")

get_mesh ¶

get_mesh() -> Optional['A2AMeshDiscovery']

Get the connected mesh discovery service.

Source code in packages/ai-parrot/src/parrot/a2a/mixin.py

def get_mesh(self) -> Optional["A2AMeshDiscovery"]:
    """Get the connected mesh discovery service."""
    return self._a2a_mesh

discover_from_mesh `async` ¶

discover_from_mesh(skill: Optional[str] = None, tag: Optional[str] = None, register_as_tools: bool = True) -> List[A2AAgentConnection]

Discover and connect to agents from the mesh.

Queries the mesh for agents matching the criteria and establishes connections to them.

PARAMETER	DESCRIPTION
`skill`	Filter by skill ID TYPE: `Optional[str]` DEFAULT: `None`
`tag`	Filter by tag TYPE: `Optional[str]` DEFAULT: `None`
`register_as_tools`	If True, register discovered agents as tools TYPE: `bool` DEFAULT: `True`

RETURNS	DESCRIPTION
`List[A2AAgentConnection]`	List of new connections established

Example

Connect to all agents with data analysis skill¶

await agent.discover_from_mesh(skill="data_analysis")

Connect to all agents tagged as "support"¶

await agent.discover_from_mesh(tag="support")

Connect to all healthy agents¶

await agent.discover_from_mesh()

Source code in packages/ai-parrot/src/parrot/a2a/mixin.py

async def discover_from_mesh(
    self,
    skill: Optional[str] = None,
    tag: Optional[str] = None,
    register_as_tools: bool = True,
) -> List[A2AAgentConnection]:
    """
    Discover and connect to agents from the mesh.

    Queries the mesh for agents matching the criteria and
    establishes connections to them.

    Args:
        skill: Filter by skill ID
        tag: Filter by tag
        register_as_tools: If True, register discovered agents as tools

    Returns:
        List of new connections established

    Example:
        # Connect to all agents with data analysis skill
        await agent.discover_from_mesh(skill="data_analysis")

        # Connect to all agents tagged as "support"
        await agent.discover_from_mesh(tag="support")

        # Connect to all healthy agents
        await agent.discover_from_mesh()
    """
    if not self._a2a_mesh:
        raise ValueError("No mesh configured. Call set_mesh() first.")

    # Query mesh based on criteria
    agents = []
    if skill:
        agents = self._a2a_mesh.get_by_skill(skill)
    elif tag:
        agents = self._a2a_mesh.get_by_tag(tag)
    else:
        agents = self._a2a_mesh.list_healthy()

    # Connect to new agents
    connections = []
    for registered_agent in agents:
        agent_name = registered_agent.card.name.lower().replace(" ", "_")
        if agent_name not in self._a2a_clients:
            # Get endpoint config for auth details
            endpoint = self._a2a_mesh.get_endpoint(registered_agent.url)
            conn = await self.add_a2a_agent(
                registered_agent.url,
                name=agent_name,
                auth_token=endpoint.auth_token if endpoint else None,
                api_key=endpoint.api_key if endpoint else None,
                headers=endpoint.headers if endpoint else None,
                register_as_tool=register_as_tools,
            )
            connections.append(conn)

    self._a2a_logger.info(
        f"Discovered {len(connections)} new agents from mesh"
    )
    return connections

set_router ¶

set_router(router: 'A2AProxyRouter') -> None

Set an A2A router for rule-based message routing.

The router enables deterministic routing based on skills, tags, or regex patterns without LLM involvement.

PARAMETER	DESCRIPTION
`router`	A2AProxyRouter instance TYPE: `'A2AProxyRouter'`

Example

router = A2AProxyRouter(mesh) router.route_by_skill("analysis", "AnalystBot") router.route_by_tag("support", "SupportBot") agent.set_router(router)

Source code in packages/ai-parrot/src/parrot/a2a/mixin.py

def set_router(self, router: "A2AProxyRouter") -> None:
    """
    Set an A2A router for rule-based message routing.

    The router enables deterministic routing based on skills,
    tags, or regex patterns without LLM involvement.

    Args:
        router: A2AProxyRouter instance

    Example:
        router = A2AProxyRouter(mesh)
        router.route_by_skill("analysis", "AnalystBot")
        router.route_by_tag("support", "SupportBot")
        agent.set_router(router)
    """
    self._a2a_router = router
    self._a2a_logger.info("Connected to A2A router")

get_router ¶

get_router() -> Optional['A2AProxyRouter']

Get the connected router.

Source code in packages/ai-parrot/src/parrot/a2a/mixin.py

def get_router(self) -> Optional["A2AProxyRouter"]:
    """Get the connected router."""
    return self._a2a_router

route_to_agent `async` ¶

route_to_agent(message: str, *, skill_id: Optional[str] = None, tags: Optional[List[str]] = None, context_id: Optional[str] = None) -> str

Route a message to an agent using the configured router.

Uses rule-based routing (no LLM) to select the appropriate agent and forward the message.

PARAMETER	DESCRIPTION
`message`	Message to route TYPE: `str`
`skill_id`	Optional skill ID hint TYPE: `Optional[str]` DEFAULT: `None`
`tags`	Optional tags hint TYPE: `Optional[List[str]]` DEFAULT: `None`
`context_id`	Optional context for multi-turn TYPE: `Optional[str]` DEFAULT: `None`

RETURNS	DESCRIPTION
`str`	Response from the routed agent

RAISES	DESCRIPTION
`ValueError`	If no router configured or no route matched

Source code in packages/ai-parrot/src/parrot/a2a/mixin.py

async def route_to_agent(
    self,
    message: str,
    *,
    skill_id: Optional[str] = None,
    tags: Optional[List[str]] = None,
    context_id: Optional[str] = None,
) -> str:
    """
    Route a message to an agent using the configured router.

    Uses rule-based routing (no LLM) to select the appropriate
    agent and forward the message.

    Args:
        message: Message to route
        skill_id: Optional skill ID hint
        tags: Optional tags hint
        context_id: Optional context for multi-turn

    Returns:
        Response from the routed agent

    Raises:
        ValueError: If no router configured or no route matched
    """
    if not self._a2a_router:
        raise ValueError("No router configured. Call set_router() first.")

    return await self._a2a_router.ask(
        message,
        skill_id=skill_id,
        tags=tags,
        context_id=context_id,
    )

set_orchestrator ¶

set_orchestrator(orchestrator: 'A2AOrchestrator') -> None

Set an A2A orchestrator for hybrid routing.

The orchestrator combines rule-based routing with LLM-driven decision making for complex scenarios.

PARAMETER	DESCRIPTION
`orchestrator`	A2AOrchestrator instance TYPE: `'A2AOrchestrator'`

Example

orchestrator = A2AOrchestrator(mesh) orchestrator.route_by_skill("simple_query", "FAQBot") orchestrator.set_fallback_llm(llm_client) agent.set_orchestrator(orchestrator)

Source code in packages/ai-parrot/src/parrot/a2a/mixin.py

def set_orchestrator(self, orchestrator: "A2AOrchestrator") -> None:
    """
    Set an A2A orchestrator for hybrid routing.

    The orchestrator combines rule-based routing with LLM-driven
    decision making for complex scenarios.

    Args:
        orchestrator: A2AOrchestrator instance

    Example:
        orchestrator = A2AOrchestrator(mesh)
        orchestrator.route_by_skill("simple_query", "FAQBot")
        orchestrator.set_fallback_llm(llm_client)
        agent.set_orchestrator(orchestrator)
    """
    self._a2a_orchestrator = orchestrator
    self._a2a_logger.info("Connected to A2A orchestrator")

get_orchestrator ¶

get_orchestrator() -> Optional['A2AOrchestrator']

Get the connected orchestrator.

Source code in packages/ai-parrot/src/parrot/a2a/mixin.py

def get_orchestrator(self) -> Optional["A2AOrchestrator"]:
    """Get the connected orchestrator."""
    return self._a2a_orchestrator

orchestrate `async` ¶

orchestrate(message: str, *, mode: Optional[str] = None, agents: Optional[List[str]] = None, context_id: Optional[str] = None) -> str

Orchestrate a message across multiple agents.

Uses the configured orchestrator which combines rules and LLM-based decision making.

PARAMETER	DESCRIPTION
`message`	Message to orchestrate TYPE: `str`
`mode`	Orchestration mode (rules, llm, hybrid, parallel, sequential) TYPE: `Optional[str]` DEFAULT: `None`
`agents`	Optional explicit list of agents TYPE: `Optional[List[str]]` DEFAULT: `None`
`context_id`	Optional context for multi-turn TYPE: `Optional[str]` DEFAULT: `None`

RETURNS	DESCRIPTION
`str`	Final response from orchestration

RAISES	DESCRIPTION
`ValueError`	If no orchestrator configured
`RuntimeError`	If orchestration fails

Source code in packages/ai-parrot/src/parrot/a2a/mixin.py

async def orchestrate(
    self,
    message: str,
    *,
    mode: Optional[str] = None,
    agents: Optional[List[str]] = None,
    context_id: Optional[str] = None,
) -> str:
    """
    Orchestrate a message across multiple agents.

    Uses the configured orchestrator which combines rules and
    LLM-based decision making.

    Args:
        message: Message to orchestrate
        mode: Orchestration mode (rules, llm, hybrid, parallel, sequential)
        agents: Optional explicit list of agents
        context_id: Optional context for multi-turn

    Returns:
        Final response from orchestration

    Raises:
        ValueError: If no orchestrator configured
        RuntimeError: If orchestration fails
    """
    if not self._a2a_orchestrator:
        raise ValueError(
            "No orchestrator configured. Call set_orchestrator() first."
        )

    # Import here to avoid circular imports
    from .orchestrator import OrchestrationMode  # pylint: disable=C0415

    orchestration_mode = None
    if mode:
        orchestration_mode = OrchestrationMode(mode)

    return await self._a2a_orchestrator.ask(
        message,
        mode=orchestration_mode,
        agents=agents,
        context_id=context_id,
    )

fan_out `async` ¶

fan_out(message: str, agents: List[str], **kwargs) -> Dict[str, Union[str, Exception]]

Send message to multiple agents in parallel.

PARAMETER	DESCRIPTION
`message`	Message to send TYPE: `str`
`agents`	List of agent names TYPE: `List[str]`
`**kwargs`	Additional arguments DEFAULT: `{}`

RETURNS	DESCRIPTION
`Dict[str, Union[str, Exception]]`	Dict mapping agent name to response or exception

Source code in packages/ai-parrot/src/parrot/a2a/mixin.py

async def fan_out(
    self,
    message: str,
    agents: List[str],
    **kwargs
) -> Dict[str, Union[str, Exception]]:
    """
    Send message to multiple agents in parallel.

    Args:
        message: Message to send
        agents: List of agent names
        **kwargs: Additional arguments

    Returns:
        Dict mapping agent name to response or exception
    """
    if not self._a2a_orchestrator:
        raise ValueError("No orchestrator configured. Call set_orchestrator() first.")

    return await self._a2a_orchestrator.fan_out(message, agents, **kwargs)

pipeline `async` ¶

pipeline(message: str, agents: List[str], **kwargs) -> str

Execute sequential pipeline across agents.

Each agent's output becomes the next agent's input.

PARAMETER	DESCRIPTION
`message`	Initial message TYPE: `str`
`agents`	Ordered list of agent names TYPE: `List[str]`
`**kwargs`	Additional arguments DEFAULT: `{}`

RETURNS	DESCRIPTION
`str`	Final output from last agent

Source code in packages/ai-parrot/src/parrot/a2a/mixin.py

async def pipeline(
    self,
    message: str,
    agents: List[str],
    **kwargs
) -> str:
    """
    Execute sequential pipeline across agents.

    Each agent's output becomes the next agent's input.

    Args:
        message: Initial message
        agents: Ordered list of agent names
        **kwargs: Additional arguments

    Returns:
        Final output from last agent
    """
    if not self._a2a_orchestrator:
        raise ValueError("No orchestrator configured. Call set_orchestrator() first.")

    return await self._a2a_orchestrator.pipeline(message, agents, **kwargs)

add_a2a_agent `async` ¶

add_a2a_agent(url: str, *, name: Optional[str] = None, auth_token: Optional[str] = None, api_key: Optional[str] = None, headers: Optional[Dict[str, str]] = None, register_as_tool: bool = True, register_skills_as_tools: bool = False, use_streaming: bool = False, timeout: float = 60.0) -> A2AAgentConnection

Connect to a remote A2A agent.

PARAMETER	DESCRIPTION
`url`	Base URL of the remote agent TYPE: `str`
`name`	Optional name override (defaults to agent's name from card) TYPE: `Optional[str]` DEFAULT: `None`
`auth_token`	Bearer token for authentication TYPE: `Optional[str]` DEFAULT: `None`
`api_key`	API key for authentication TYPE: `Optional[str]` DEFAULT: `None`
`headers`	Additional headers TYPE: `Optional[Dict[str, str]]` DEFAULT: `None`
`register_as_tool`	If True, register the agent as a callable tool TYPE: `bool` DEFAULT: `True`
`register_skills_as_tools`	If True, register each skill as a separate tool TYPE: `bool` DEFAULT: `False`
`use_streaming`	If True, use streaming for tool calls TYPE: `bool` DEFAULT: `False`
`timeout`	Request timeout TYPE: `float` DEFAULT: `60.0`

RETURNS	DESCRIPTION
`A2AAgentConnection`	A2AAgentConnection with the remote agent info

Source code in packages/ai-parrot/src/parrot/a2a/mixin.py

async def add_a2a_agent(
    self,
    url: str,
    *,
    name: Optional[str] = None,
    auth_token: Optional[str] = None,
    api_key: Optional[str] = None,
    headers: Optional[Dict[str, str]] = None,
    register_as_tool: bool = True,
    register_skills_as_tools: bool = False,
    use_streaming: bool = False,
    timeout: float = 60.0,
) -> A2AAgentConnection:
    """
    Connect to a remote A2A agent.

    Args:
        url: Base URL of the remote agent
        name: Optional name override (defaults to agent's name from card)
        auth_token: Bearer token for authentication
        api_key: API key for authentication
        headers: Additional headers
        register_as_tool: If True, register the agent as a callable tool
        register_skills_as_tools: If True, register each skill as a separate tool
        use_streaming: If True, use streaming for tool calls
        timeout: Request timeout

    Returns:
        A2AAgentConnection with the remote agent info
    """
    client = A2AClient(
        url,
        auth_token=auth_token,
        api_key=api_key,
        headers=headers,
        timeout=timeout,
    )

    await client.connect()

    card = client.agent_card
    agent_name = name or card.name.lower().replace(" ", "_")

    connection = A2AAgentConnection(
        url=url,
        card=card,
        client=client,
        name=agent_name,
    )

    self._a2a_clients[agent_name] = connection

    # Register as tool(s)
    if register_as_tool:
        tool = A2ARemoteAgentTool(
            client,
            tool_name=f"ask_{agent_name}",
            use_streaming=use_streaming,
        )
        self._register_a2a_tool(tool)
        self._a2a_logger.info(
            f"Registered remote agent '{agent_name}' as tool: {tool.name}"
        )

    if register_skills_as_tools:
        for skill in card.skills:
            skill_tool = A2ARemoteSkillTool(client, skill)
            self._register_a2a_tool(skill_tool)
            self._a2a_logger.info(
                f"Registered remote skill as tool: {skill_tool.name}"
            )

    self._a2a_logger.info(
        f"Connected to A2A agent '{card.name}' at {url} "
        f"with {len(card.skills)} skills"
    )

    return connection

remove_a2a_agent `async` ¶

remove_a2a_agent(name: str) -> None

Disconnect from a remote A2A agent.

PARAMETER	DESCRIPTION
`name`	Name of the agent to disconnect TYPE: `str`

Source code in packages/ai-parrot/src/parrot/a2a/mixin.py

async def remove_a2a_agent(self, name: str) -> None:
    """
    Disconnect from a remote A2A agent.

    Args:
        name: Name of the agent to disconnect
    """
    if name not in self._a2a_clients:
        self._a2a_logger.warning(f"A2A agent '{name}' not found")
        return

    connection = self._a2a_clients[name]

    # Remove tools
    if hasattr(self, 'tool_manager') and self.tool_manager:
        tool_name = f"ask_{name}"
        if self.tool_manager.get_tool(tool_name):
            self.tool_manager.unregister_tool(tool_name)

        # Remove skill tools
        for skill in connection.card.skills:
            skill_tool_name = f"remote_{skill.id}"
            if self.tool_manager.get_tool(skill_tool_name):
                self.tool_manager.unregister_tool(skill_tool_name)

    await connection.client.disconnect()
    del self._a2a_clients[name]

    self._a2a_logger.info(f"Disconnected from A2A agent '{name}'")

list_a2a_agents ¶

list_a2a_agents() -> List[str]

List connected A2A agent names.

Source code in packages/ai-parrot/src/parrot/a2a/mixin.py

def list_a2a_agents(self) -> List[str]:
    """List connected A2A agent names."""
    return list(self._a2a_clients.keys())

get_a2a_agent ¶

get_a2a_agent(name: str) -> Optional[A2AAgentConnection]

Get a connected A2A agent by name.

Source code in packages/ai-parrot/src/parrot/a2a/mixin.py

def get_a2a_agent(self, name: str) -> Optional[A2AAgentConnection]:
    """Get a connected A2A agent by name."""
    return self._a2a_clients.get(name)

get_a2a_client ¶

get_a2a_client(name: str) -> Optional[A2AClient]

Get the A2A client for a connected agent.

Source code in packages/ai-parrot/src/parrot/a2a/mixin.py

def get_a2a_client(self, name: str) -> Optional[A2AClient]:
    """Get the A2A client for a connected agent."""
    conn = self._a2a_clients.get(name)
    return conn.client if conn else None

ask_remote_agent `async` ¶

ask_remote_agent(agent_name: str, question: str, *, context_id: Optional[str] = None, stream: bool = False) -> str

Ask a question directly to a remote A2A agent.

PARAMETER	DESCRIPTION
`agent_name`	Name of the connected agent TYPE: `str`
`question`	The question to ask TYPE: `str`
`context_id`	Optional context for multi-turn TYPE: `Optional[str]` DEFAULT: `None`
`stream`	If True, stream the response TYPE: `bool` DEFAULT: `False`

RETURNS	DESCRIPTION
`str`	The agent's response as text

RAISES	DESCRIPTION
`ValueError`	If agent not connected
`RuntimeError`	If agent returns error

Source code in packages/ai-parrot/src/parrot/a2a/mixin.py

async def ask_remote_agent(
    self,
    agent_name: str,
    question: str,
    *,
    context_id: Optional[str] = None,
    stream: bool = False,
) -> str:
    """
    Ask a question directly to a remote A2A agent.

    Args:
        agent_name: Name of the connected agent
        question: The question to ask
        context_id: Optional context for multi-turn
        stream: If True, stream the response

    Returns:
        The agent's response as text

    Raises:
        ValueError: If agent not connected
        RuntimeError: If agent returns error
    """
    conn = self._a2a_clients.get(agent_name)
    if not conn:
        raise ValueError(f"A2A agent '{agent_name}' not connected")

    if stream:
        chunks = []
        async for chunk in conn.client.stream_message(
            question, context_id=context_id
        ):
            chunks.append(chunk)
        return "".join(chunks)
    else:
        task = await conn.client.send_message(question, context_id=context_id)

        if task.status.state == TaskState.FAILED:
            error = (
                task.status.message.get_text()
                if task.status.message
                else "Unknown error"
            )
            raise RuntimeError(f"Remote agent error: {error}")

        if task.artifacts and task.artifacts[0].parts:
            return task.artifacts[0].parts[0].text or ""
        return ""

invoke_remote_skill `async` ¶

invoke_remote_skill(agent_name: str, skill_id: str, params: Optional[Dict[str, Any]] = None, *, context_id: Optional[str] = None) -> Any

Invoke a specific skill on a remote agent.

PARAMETER	DESCRIPTION
`agent_name`	Name of the connected agent TYPE: `str`
`skill_id`	ID of the skill to invoke TYPE: `str`
`params`	Parameters for the skill TYPE: `Optional[Dict[str, Any]]` DEFAULT: `None`
`context_id`	Optional context TYPE: `Optional[str]` DEFAULT: `None`

RETURNS	DESCRIPTION
`Any`	The skill result

RAISES	DESCRIPTION
`ValueError`	If agent not connected

Source code in packages/ai-parrot/src/parrot/a2a/mixin.py

async def invoke_remote_skill(
    self,
    agent_name: str,
    skill_id: str,
    params: Optional[Dict[str, Any]] = None,
    *,
    context_id: Optional[str] = None,
) -> Any:
    """
    Invoke a specific skill on a remote agent.

    Args:
        agent_name: Name of the connected agent
        skill_id: ID of the skill to invoke
        params: Parameters for the skill
        context_id: Optional context

    Returns:
        The skill result

    Raises:
        ValueError: If agent not connected
    """
    conn = self._a2a_clients.get(agent_name)
    if not conn:
        raise ValueError(f"A2A agent '{agent_name}' not connected")

    return await conn.client.invoke_skill(
        skill_id, params, context_id=context_id
    )

shutdown_a2a `async` ¶

shutdown_a2a() -> None

Disconnect all A2A connections and cleanup resources.

Source code in packages/ai-parrot/src/parrot/a2a/mixin.py

async def shutdown_a2a(self) -> None:
    """Disconnect all A2A connections and cleanup resources."""
    # Disconnect direct connections
    for name in list(self._a2a_clients.keys()):
        await self.remove_a2a_agent(name)

    # Close router clients if present
    if self._a2a_router:
        await self._a2a_router.close_clients()

    # Close orchestrator clients if present
    if self._a2a_orchestrator:
        await self._a2a_orchestrator.close_clients()

    self._a2a_logger.info("A2A connections shutdown complete")

shutdown `async` ¶

shutdown(**kwargs) -> None

Override shutdown to cleanup A2A connections.

Source code in packages/ai-parrot/src/parrot/a2a/mixin.py

async def shutdown(self, **kwargs) -> None:
    """Override shutdown to cleanup A2A connections."""
    await self.shutdown_a2a()

    if hasattr(super(), 'shutdown'):
        await super().shutdown(**kwargs)

AgentCard `dataclass` ¶

AgentCard(name: str, description: str, version: str, skills: List[AgentSkill], supported_interfaces: List[AgentInterface] = list(), capabilities: AgentCapabilities = AgentCapabilities(), default_input_modes: List[str] = (lambda: ['text/plain', 'application/json'])(), default_output_modes: List[str] = (lambda: ['text/plain', 'application/json'])(), provider: Optional[AgentProvider] = None, documentation_url: Optional[str] = None, security_schemes: Optional[Dict[str, SecurityScheme]] = None, security_requirements: Optional[List[SecurityRequirement]] = None, signatures: Optional[List[AgentCardSignature]] = None, icon_url: Optional[str] = None, tags: List[str] = list())

Self-describing manifest for an agent (A2A v1.0 structure).

Replaces the flat v0.3 url + preferredTransport with a structured supported_interfaces array. The flat accessors remain available as read-only backward-compat properties (url, preferred_transport, protocol_version) so existing consumers keep working.

url `property` `writable` ¶

url: Optional[str]

Backward-compat: first interface URL (the v0.3 flat url).

preferred_transport `property` `writable` ¶

preferred_transport: str

Backward-compat: first interface protocol binding.

protocol_version `property` ¶

protocol_version: str

Backward-compat: first interface protocol version.

AgentSkill `dataclass` ¶

AgentSkill(id: str, name: str, description: str, tags: List[str] = list(), input_schema: Optional[Dict[str, Any]] = None, examples: List[str] = list(), input_modes: Optional[List[str]] = None, output_modes: Optional[List[str]] = None, security_requirements: Optional[List[SecurityRequirement]] = None)

A capability exposed by an agent (maps to a tool).

AgentCapabilities `dataclass` ¶

AgentCapabilities(streaming: bool = True, push_notifications: bool = False, extended_agent_card: bool = False, extensions: List[AgentExtension] = list())

Capabilities supported by an agent.

Task `dataclass` ¶

Task(id: str, context_id: str, status: TaskStatus, artifacts: List[Artifact] = list(), history: List[Message] = list(), metadata: Optional[Dict[str, Any]] = None)

Unit of work with lifecycle.

TaskState ¶

Bases: str, Enum

Task lifecycle states — v1.0.0 ProtoJSON values.

The canonical member value is the v1.0 TASK_STATE_* string. The legacy v0.3 lowercase values ("submitted" …) are handled by :func:parse_task_state on deserialization and :func:serialize_task_state on serialization.

TaskStatus `dataclass` ¶

TaskStatus(state: TaskState, message: Optional[Message] = None, timestamp: Optional[str] = None)

Current status of a task.

Message `dataclass` ¶

Message(message_id: str, role: Role, parts: List[Part], context_id: Optional[str] = None, task_id: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None, extensions: Optional[List[str]] = None, reference_task_ids: Optional[List[str]] = None)

Communication unit between agents.

get_text ¶

get_text() -> str

Extract all text content from parts.

Source code in packages/ai-parrot/src/parrot/a2a/models.py

def get_text(self) -> str:
    """Extract all text content from parts."""
    return " ".join(p.text for p in self.parts if p.text)

get_data ¶

get_data() -> Optional[Dict[str, Any]]

Extract structured data from parts.

Source code in packages/ai-parrot/src/parrot/a2a/models.py

def get_data(self) -> Optional[Dict[str, Any]]:
    """Extract structured data from parts."""
    return next((p.data for p in self.parts if p.data), None)

Part `dataclass` ¶

Part(text: Optional[str] = None, file_uri: Optional[str] = None, file_bytes: Optional[bytes] = None, file_media_type: Optional[str] = None, filename: Optional[str] = None, data: Optional[Dict[str, Any]] = None, metadata: Optional[Dict[str, Any]] = None)

Atomic content unit.

Artifact `dataclass` ¶

Artifact(artifact_id: str, parts: List[Part], name: Optional[str] = None, description: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None)

Output produced by an agent.

from_response `classmethod` ¶

from_response(response: Any, name: str = 'response') -> Artifact

Create artifact from an AIMessage or string response.

Source code in packages/ai-parrot/src/parrot/a2a/models.py

@classmethod
def from_response(cls, response: Any, name: str = "response") -> "Artifact":
    """Create artifact from an AIMessage or string response."""
    if hasattr(response, 'content'):
        # AIMessage
        text = response.content
    elif hasattr(response, 'response'):
        text = response.response
    else:
        text = str(response)

    return cls(
        artifact_id=str(uuid.uuid4()),
        name=name,
        parts=[Part.from_text(text)]
    )

RegisteredAgent `dataclass` ¶

RegisteredAgent(url: str, card: AgentCard, last_seen: datetime = datetime.utcnow(), healthy: bool = True, protocol_version: str = '0.3')

Definition about a Registered Agent.

AgentConfig `dataclass` ¶

AgentConfig(name: str, description: str, port: int, skills: List[Dict[str, Any]], system_prompt: str)

Configuration for an A2A agent.

AgentInterface `dataclass` ¶

AgentInterface(url: str, protocol_binding: str, protocol_version: str = '1.0', tenant: Optional[str] = None)

v1.0 AgentCard interface entry.

AgentProvider `dataclass` ¶

AgentProvider(url: str, organization: str)

Organization that provides the agent (v1.0).

AgentExtension `dataclass` ¶

AgentExtension(uri: str, description: Optional[str] = None, required: bool = False, params: Optional[Dict[str, Any]] = None)

A protocol extension declared by an agent (v1.0).

AgentCardSignature `dataclass` ¶

AgentCardSignature(protected: str, signature: str, header: Optional[Dict[str, Any]] = None)

A JWS signature over the AgentCard (v1.0). Signing itself is out of scope.

SecurityScheme `dataclass` ¶

SecurityScheme(type: str, description: Optional[str] = None)

Base security scheme (v1.0 securitySchemes entry).

APIKeySecurityScheme `dataclass` ¶

APIKeySecurityScheme(name: str = '', location: str = 'header', description: Optional[str] = None)

Bases: SecurityScheme

API key security scheme.

Source code in packages/ai-parrot/src/parrot/a2a/models.py

def __init__(self, name: str = "", location: str = "header",
             description: Optional[str] = None):
    super().__init__(type="apiKey", description=description)
    self.name = name
    self.location = location

HTTPAuthSecurityScheme `dataclass` ¶

HTTPAuthSecurityScheme(scheme: str = 'bearer', bearer_format: Optional[str] = None, description: Optional[str] = None)

Bases: SecurityScheme

HTTP authentication security scheme (Bearer/Basic).

Source code in packages/ai-parrot/src/parrot/a2a/models.py

def __init__(self, scheme: str = "bearer", bearer_format: Optional[str] = None,
             description: Optional[str] = None):
    super().__init__(type="http", description=description)
    self.scheme = scheme
    self.bearer_format = bearer_format

OAuth2SecurityScheme `dataclass` ¶

OAuth2SecurityScheme(flows: Optional[Dict[str, Any]] = None, description: Optional[str] = None)

Bases: SecurityScheme

OAuth 2.0 security scheme.

Source code in packages/ai-parrot/src/parrot/a2a/models.py

def __init__(self, flows: Optional[Dict[str, Any]] = None,
             description: Optional[str] = None):
    super().__init__(type="oauth2", description=description)
    self.flows = flows or {}

OpenIdConnectSecurityScheme `dataclass` ¶

OpenIdConnectSecurityScheme(open_id_connect_url: str = '', description: Optional[str] = None)

Bases: SecurityScheme

OpenID Connect security scheme.

Source code in packages/ai-parrot/src/parrot/a2a/models.py

def __init__(self, open_id_connect_url: str = "",
             description: Optional[str] = None):
    super().__init__(type="openIdConnect", description=description)
    self.open_id_connect_url = open_id_connect_url

MutualTlsSecurityScheme `dataclass` ¶

MutualTlsSecurityScheme(description: Optional[str] = None)

Bases: SecurityScheme

Mutual TLS security scheme.

Source code in packages/ai-parrot/src/parrot/a2a/models.py

def __init__(self, description: Optional[str] = None):
    super().__init__(type="mutualTLS", description=description)

SecurityRequirement `dataclass` ¶

SecurityRequirement(schemes: Dict[str, List[str]] = dict())

A security requirement: a map of scheme name -> required scopes.

SendMessageConfiguration `dataclass` ¶

SendMessageConfiguration(accepted_output_modes: Optional[List[str]] = None, task_push_notification_config: Optional[TaskPushNotificationConfig] = None, history_length: Optional[int] = None, return_immediately: bool = False)

Configuration accompanying a SendMessage request (v1.0).

TaskPushNotificationConfig `dataclass` ¶

TaskPushNotificationConfig(id: str, task_id: str, url: str, authentication: Optional[AuthenticationInfo] = None, metadata: Optional[Dict[str, Any]] = None)

Configuration for a task's push-notification webhook (v1.0).

AuthenticationInfo `dataclass` ¶

AuthenticationInfo(scheme: str, credentials: Optional[str] = None)

Authentication details for a push notification webhook (v1.0).

A2AError `dataclass` ¶

A2AError(code: int, message: str, data: Optional[Any] = None)

A2A JSON-RPC error object.

Role ¶

Bases: str, Enum

Message role — v1.0.0 ProtoJSON values.

A2AMeshDiscovery ¶

A2AMeshDiscovery(health_check_interval: float = 30.0, health_check_timeout: float = 10.0, auto_discover_on_start: bool = True, max_concurrent_health_checks: int = 10, retry_unhealthy_interval: float = 60.0, remove_after_failures: int = 0)

Centralized discovery service for remote A2A agents.

Provides a registry of remote A2A agents with automatic health checking, multiple lookup methods, and event notifications.

Features

Register agents by URL with automatic card discovery
Periodic health checks with configurable intervals
Lookup by name, skill ID, skill name, or tag
Full-text search across agent metadata
Event callbacks for status changes
YAML configuration with environment variable substitution
Statistics and monitoring

Example

Basic usage¶

mesh = A2AMeshDiscovery(health_check_interval=60.0) await mesh.start()

agent_card = await mesh.register("http://my-agent:8080") print(f"Registered: {agent_card.name}")

Query by skill¶

analysts = mesh.get_by_skill("data_analysis") for agent in analysts: print(f" - {agent.card.name} at {agent.url}")

await mesh.stop()

From config file¶

mesh = A2AMeshDiscovery.from_config("config/a2a_agents.yaml") await mesh.start() # Discovers all configured endpoints

Initialize the mesh discovery service.

PARAMETER	DESCRIPTION
`health_check_interval`	Seconds between health checks for healthy agents TYPE: `float` DEFAULT: `30.0`
`health_check_timeout`	Timeout for health check requests TYPE: `float` DEFAULT: `10.0`
`auto_discover_on_start`	If True, discover all endpoints on start() TYPE: `bool` DEFAULT: `True`
`max_concurrent_health_checks`	Max concurrent health check requests TYPE: `int` DEFAULT: `10`
`retry_unhealthy_interval`	Seconds between retries for unhealthy agents TYPE: `float` DEFAULT: `60.0`
`remove_after_failures`	Remove agent after N consecutive failures (0 = never) TYPE: `int` DEFAULT: `0`

Source code in packages/ai-parrot/src/parrot/a2a/mesh.py

def __init__(
    self,
    health_check_interval: float = 30.0,
    health_check_timeout: float = 10.0,
    auto_discover_on_start: bool = True,
    max_concurrent_health_checks: int = 10,
    retry_unhealthy_interval: float = 60.0,
    remove_after_failures: int = 0,  # 0 = never remove
):
    """
    Initialize the mesh discovery service.

    Args:
        health_check_interval: Seconds between health checks for healthy agents
        health_check_timeout: Timeout for health check requests
        auto_discover_on_start: If True, discover all endpoints on start()
        max_concurrent_health_checks: Max concurrent health check requests
        retry_unhealthy_interval: Seconds between retries for unhealthy agents
        remove_after_failures: Remove agent after N consecutive failures (0 = never)
    """
    self._health_check_interval = health_check_interval
    self._health_check_timeout = health_check_timeout
    self._auto_discover_on_start = auto_discover_on_start
    self._max_concurrent_health_checks = max_concurrent_health_checks
    self._retry_unhealthy_interval = retry_unhealthy_interval
    self._remove_after_failures = remove_after_failures

    # Storage
    self._endpoints: Dict[str, A2AEndpoint] = {}      # url -> endpoint config
    self._discovered: Dict[str, RegisteredAgent] = {}  # name -> discovered agent
    self._url_to_name: Dict[str, str] = {}             # url -> agent name (reverse lookup)
    self._failure_counts: Dict[str, int] = {}          # name -> consecutive failures

    # Runtime state
    self._health_task: Optional[asyncio.Task] = None
    self._running: bool = False
    self._stats = DiscoveryStats()
    self._semaphore: Optional[asyncio.Semaphore] = None

    # Event callbacks
    self._on_agent_healthy: List[AgentEventCallback] = []
    self._on_agent_unhealthy: List[AgentEventCallback] = []
    self._on_agent_registered: List[AgentEventCallback] = []
    self._on_agent_removed: List[AgentEventCallback] = []

    self.logger = logging.getLogger("Parrot.A2AMesh")

stats `property` ¶

stats: DiscoveryStats

Get current discovery statistics.

from_config `classmethod` ¶

from_config(config_path: Union[str, Path], **kwargs) -> 'A2AMeshDiscovery'

Create mesh from YAML configuration file.

The YAML file supports environment variable substitution using ${VAR_NAME}.

Example YAML

settings: health_check_interval: 30 health_check_timeout: 10 auto_discover_on_start: true

agents: - url: http://sales-agent:8080 tags: [sales, revenue]

url: http://support-agent:8080 auth_token: ${SUPPORT_AGENT_TOKEN} tags: [support, customer]
url: https://external-agent.example.com api_key: ${EXTERNAL_API_KEY} timeout: 60 enabled: true

PARAMETER	DESCRIPTION
`config_path`	Path to YAML configuration file TYPE: `Union[str, Path]`
`**kwargs`	Override settings from config DEFAULT: `{}`

RETURNS	DESCRIPTION
`'A2AMeshDiscovery'`	Configured A2AMeshDiscovery instance

RAISES	DESCRIPTION
`FileNotFoundError`	If config file doesn't exist
`ValueError`	If config format is invalid

Source code in packages/ai-parrot/src/parrot/a2a/mesh.py

@classmethod
def from_config(
    cls,
    config_path: Union[str, Path],
    **kwargs
) -> "A2AMeshDiscovery":
    """
    Create mesh from YAML configuration file.

    The YAML file supports environment variable substitution using ${VAR_NAME}.

    Example YAML:
        settings:
          health_check_interval: 30
          health_check_timeout: 10
          auto_discover_on_start: true

        agents:
          - url: http://sales-agent:8080
            tags: [sales, revenue]

          - url: http://support-agent:8080
            auth_token: ${SUPPORT_AGENT_TOKEN}
            tags: [support, customer]

          - url: https://external-agent.example.com
            api_key: ${EXTERNAL_API_KEY}
            timeout: 60
            enabled: true

    Args:
        config_path: Path to YAML configuration file
        **kwargs: Override settings from config

    Returns:
        Configured A2AMeshDiscovery instance

    Raises:
        FileNotFoundError: If config file doesn't exist
        ValueError: If config format is invalid
    """
    path = Path(config_path)
    if not path.exists():
        raise FileNotFoundError(f"Config file not found: {path}")

    with open(path, "r", encoding="utf-8") as f:
        raw_config = yaml.safe_load(f)

    if not raw_config:
        raw_config = {}

    # Process environment variables
    config = cls._substitute_env_vars(raw_config)

    # Extract settings
    settings = config.get("settings", {})
    merged_settings = {
        "health_check_interval": settings.get("health_check_interval", 30.0),
        "health_check_timeout": settings.get("health_check_timeout", 10.0),
        "auto_discover_on_start": settings.get("auto_discover_on_start", True),
        "max_concurrent_health_checks": settings.get("max_concurrent_health_checks", 10),
        "retry_unhealthy_interval": settings.get("retry_unhealthy_interval", 60.0),
        "remove_after_failures": settings.get("remove_after_failures", 0),
    }
    merged_settings |= kwargs

    # Create instance
    instance = cls(**merged_settings)

    # Add endpoints from config
    agents_config = config.get("agents", [])
    for agent_config in agents_config:
        if isinstance(agent_config, str):
            # Simple URL string
            instance.add_endpoint(agent_config)
        elif isinstance(agent_config, dict):
            # Full endpoint configuration
            if url := agent_config.pop("url", None):
                instance.add_endpoint(url, **agent_config)

    instance.logger.info(
        f"Loaded {len(instance._endpoints)} endpoints from {path}"
    )

    return instance

start `async` ¶

start() -> None

Start the mesh discovery service.

This will: 1. Initialize the concurrency semaphore 2. Discover all configured endpoints (if auto_discover_on_start is True) 3. Start the background health check loop

Source code in packages/ai-parrot/src/parrot/a2a/mesh.py

async def start(self) -> None:
    """
    Start the mesh discovery service.

    This will:
    1. Initialize the concurrency semaphore
    2. Discover all configured endpoints (if auto_discover_on_start is True)
    3. Start the background health check loop
    """
    if self._running:
        self.logger.warning("Mesh discovery already running")
        return

    self._running = True
    self._semaphore = asyncio.Semaphore(self._max_concurrent_health_checks)

    self.logger.info(
        f"Starting A2A Mesh Discovery with {len(self._endpoints)} endpoints"
    )

    # Auto-discover configured endpoints
    if self._auto_discover_on_start and self._endpoints:
        await self._discover_all_endpoints()

    # Start health check loop
    self._health_task = asyncio.create_task(
        self._health_check_loop(),
        name="a2a_mesh_health_check"
    )

    self.logger.info(
        f"A2A Mesh Discovery started: "
        f"{self._stats.total_healthy}/{self._stats.total_registered} agents healthy"
    )

stop `async` ¶

stop() -> None

Stop the mesh discovery service.

Cancels the health check loop and cleans up resources.

Source code in packages/ai-parrot/src/parrot/a2a/mesh.py

async def stop(self) -> None:
    """
    Stop the mesh discovery service.

    Cancels the health check loop and cleans up resources.
    """
    if not self._running:
        return

    self._running = False

    if self._health_task:
        self._health_task.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await self._health_task
        self._health_task = None

    self.logger.info("A2A Mesh Discovery stopped")

add_endpoint ¶

add_endpoint(url: str, *, name: Optional[str] = None, auth_token: Optional[str] = None, api_key: Optional[str] = None, headers: Optional[Dict[str, str]] = None, tags: Optional[Union[Set[str], List[str]]] = None, timeout: float = 30.0, health_check_strategy: Union[HealthCheckStrategy, str] = HealthCheckStrategy.DISCOVERY, health_check_endpoint: Optional[str] = None, enabled: bool = True, priority: int = 0, metadata: Optional[Dict[str, Any]] = None) -> 'A2AMeshDiscovery'

Add an endpoint configuration for later discovery.

The endpoint won't be discovered until start() is called or register() is called explicitly.

PARAMETER	DESCRIPTION
`url`	Base URL of the A2A agent TYPE: `str`
`name`	Optional name hint TYPE: `Optional[str]` DEFAULT: `None`
`auth_token`	Bearer token for authentication TYPE: `Optional[str]` DEFAULT: `None`
`api_key`	API key for X-API-Key header TYPE: `Optional[str]` DEFAULT: `None`
`headers`	Additional HTTP headers TYPE: `Optional[Dict[str, str]]` DEFAULT: `None`
`tags`	Tags for categorization TYPE: `Optional[Union[Set[str], List[str]]]` DEFAULT: `None`
`timeout`	Request timeout TYPE: `float` DEFAULT: `30.0`
`health_check_strategy`	Strategy for health checks TYPE: `Union[HealthCheckStrategy, str]` DEFAULT: `DISCOVERY`
`health_check_endpoint`	Custom health check endpoint TYPE: `Optional[str]` DEFAULT: `None`
`enabled`	Whether endpoint is enabled TYPE: `bool` DEFAULT: `True`
`priority`	Priority for load balancing TYPE: `int` DEFAULT: `0`
`metadata`	Additional metadata TYPE: `Optional[Dict[str, Any]]` DEFAULT: `None`

RETURNS	DESCRIPTION
`'A2AMeshDiscovery'`	Self for method chaining

Source code in packages/ai-parrot/src/parrot/a2a/mesh.py

def add_endpoint(
    self,
    url: str,
    *,
    name: Optional[str] = None,
    auth_token: Optional[str] = None,
    api_key: Optional[str] = None,
    headers: Optional[Dict[str, str]] = None,
    tags: Optional[Union[Set[str], List[str]]] = None,
    timeout: float = 30.0,
    health_check_strategy: Union[HealthCheckStrategy, str] = HealthCheckStrategy.DISCOVERY,
    health_check_endpoint: Optional[str] = None,
    enabled: bool = True,
    priority: int = 0,
    metadata: Optional[Dict[str, Any]] = None,
) -> "A2AMeshDiscovery":
    """
    Add an endpoint configuration for later discovery.

    The endpoint won't be discovered until start() is called or
    register() is called explicitly.

    Args:
        url: Base URL of the A2A agent
        name: Optional name hint
        auth_token: Bearer token for authentication
        api_key: API key for X-API-Key header
        headers: Additional HTTP headers
        tags: Tags for categorization
        timeout: Request timeout
        health_check_strategy: Strategy for health checks
        health_check_endpoint: Custom health check endpoint
        enabled: Whether endpoint is enabled
        priority: Priority for load balancing
        metadata: Additional metadata

    Returns:
        Self for method chaining
    """
    url = url.rstrip("/")

    if isinstance(health_check_strategy, str):
        health_check_strategy = HealthCheckStrategy(health_check_strategy)

    endpoint = A2AEndpoint(
        url=url,
        name=name,
        auth_token=auth_token,
        api_key=api_key,
        headers=headers,
        tags=set(tags) if tags else set(),
        timeout=timeout,
        health_check_strategy=health_check_strategy,
        health_check_endpoint=health_check_endpoint,
        enabled=enabled,
        priority=priority,
        metadata=metadata or {},
    )

    self._endpoints[url] = endpoint
    self.logger.debug(f"Added endpoint: {url}")

    return self

remove_endpoint ¶

remove_endpoint(url: str) -> bool

Remove an endpoint configuration.

Also removes the discovered agent if present.

PARAMETER	DESCRIPTION
`url`	URL of the endpoint to remove TYPE: `str`

RETURNS	DESCRIPTION
`bool`	True if endpoint was removed, False if not found

Source code in packages/ai-parrot/src/parrot/a2a/mesh.py

def remove_endpoint(self, url: str) -> bool:
    """
    Remove an endpoint configuration.

    Also removes the discovered agent if present.

    Args:
        url: URL of the endpoint to remove

    Returns:
        True if endpoint was removed, False if not found
    """
    url = url.rstrip("/")

    if url not in self._endpoints:
        return False

    del self._endpoints[url]

    # Remove discovered agent if present
    if url in self._url_to_name:
        name = self._url_to_name[url]
        if name in self._discovered:
            agent = self._discovered.pop(name)
            del self._url_to_name[url]
            self._failure_counts.pop(name, None)
            self._update_stats()
            asyncio.create_task(self._emit_event("removed", agent))

    self.logger.debug(f"Removed endpoint: {url}")
    return True

get_endpoint ¶

get_endpoint(url: str) -> Optional[A2AEndpoint]

Get endpoint configuration by URL.

Source code in packages/ai-parrot/src/parrot/a2a/mesh.py

def get_endpoint(self, url: str) -> Optional[A2AEndpoint]:
    """Get endpoint configuration by URL."""
    return self._endpoints.get(url.rstrip("/"))

list_endpoints ¶

list_endpoints() -> List[A2AEndpoint]

List all configured endpoints.

Source code in packages/ai-parrot/src/parrot/a2a/mesh.py

def list_endpoints(self) -> List[A2AEndpoint]:
    """List all configured endpoints."""
    return list(self._endpoints.values())

register `async` ¶

register(url: str, *, auth_token: Optional[str] = None, api_key: Optional[str] = None, headers: Optional[Dict[str, str]] = None, tags: Optional[Union[Set[str], List[str]]] = None, timeout: float = 30.0, **kwargs) -> RegisteredAgent

Register and discover an agent immediately.

Connects to the agent, fetches its AgentCard, and adds it to the registry. This method both configures the endpoint and performs discovery.

PARAMETER	DESCRIPTION
`url`	Base URL of the A2A agent TYPE: `str`
`auth_token`	Bearer token for authentication TYPE: `Optional[str]` DEFAULT: `None`
`api_key`	API key for X-API-Key header TYPE: `Optional[str]` DEFAULT: `None`
`headers`	Additional HTTP headers TYPE: `Optional[Dict[str, str]]` DEFAULT: `None`
`tags`	Additional tags to merge with agent's tags TYPE: `Optional[Union[Set[str], List[str]]]` DEFAULT: `None`
`timeout`	Request timeout TYPE: `float` DEFAULT: `30.0`
`**kwargs`	Additional endpoint configuration DEFAULT: `{}`

RETURNS	DESCRIPTION
`RegisteredAgent`	RegisteredAgent with discovered card

RAISES	DESCRIPTION
`ConnectionError`	If agent is unreachable
`ValueError`	If agent card is invalid

Source code in packages/ai-parrot/src/parrot/a2a/mesh.py

async def register(
    self,
    url: str,
    *,
    auth_token: Optional[str] = None,
    api_key: Optional[str] = None,
    headers: Optional[Dict[str, str]] = None,
    tags: Optional[Union[Set[str], List[str]]] = None,
    timeout: float = 30.0,
    **kwargs
) -> RegisteredAgent:
    """
    Register and discover an agent immediately.

    Connects to the agent, fetches its AgentCard, and adds it to the registry.
    This method both configures the endpoint and performs discovery.

    Args:
        url: Base URL of the A2A agent
        auth_token: Bearer token for authentication
        api_key: API key for X-API-Key header
        headers: Additional HTTP headers
        tags: Additional tags to merge with agent's tags
        timeout: Request timeout
        **kwargs: Additional endpoint configuration

    Returns:
        RegisteredAgent with discovered card

    Raises:
        ConnectionError: If agent is unreachable
        ValueError: If agent card is invalid
    """
    url = url.rstrip("/")

    # Add to endpoints if not already configured
    if url not in self._endpoints:
        self.add_endpoint(
            url,
            auth_token=auth_token,
            api_key=api_key,
            headers=headers,
            tags=tags,
            timeout=timeout,
            **kwargs
        )

    endpoint = self._endpoints[url]

    # Perform discovery
    return await self._discover_endpoint(endpoint)

unregister `async` ¶

unregister(name: str) -> bool

Unregister an agent by name.

Removes the agent from the registry but keeps the endpoint configuration.

PARAMETER	DESCRIPTION
`name`	Name of the agent to unregister TYPE: `str`

RETURNS	DESCRIPTION
`bool`	True if agent was unregistered, False if not found

Source code in packages/ai-parrot/src/parrot/a2a/mesh.py

async def unregister(self, name: str) -> bool:
    """
    Unregister an agent by name.

    Removes the agent from the registry but keeps the endpoint configuration.

    Args:
        name: Name of the agent to unregister

    Returns:
        True if agent was unregistered, False if not found
    """
    if name not in self._discovered:
        return False

    agent = self._discovered.pop(name)

    # Clean up reverse lookup
    for url, agent_name in list(self._url_to_name.items()):
        if agent_name == name:
            del self._url_to_name[url]
            break

    self._failure_counts.pop(name, None)
    self._update_stats()

    await self._emit_event("removed", agent)

    self.logger.info(f"Unregistered agent: {name}")
    return True

get ¶

get(name: str) -> Optional[RegisteredAgent]

Get a registered agent by name.

PARAMETER	DESCRIPTION
`name`	Exact name of the agent TYPE: `str`

RETURNS	DESCRIPTION
`Optional[RegisteredAgent]`	RegisteredAgent if found, None otherwise

Source code in packages/ai-parrot/src/parrot/a2a/mesh.py

def get(self, name: str) -> Optional[RegisteredAgent]:
    """
    Get a registered agent by name.

    Args:
        name: Exact name of the agent

    Returns:
        RegisteredAgent if found, None otherwise
    """
    return self._discovered.get(name)

get_by_url ¶

get_by_url(url: str) -> Optional[RegisteredAgent]

Get a registered agent by URL.

PARAMETER	DESCRIPTION
`url`	URL of the agent TYPE: `str`

RETURNS	DESCRIPTION
`Optional[RegisteredAgent]`	RegisteredAgent if found, None otherwise

Source code in packages/ai-parrot/src/parrot/a2a/mesh.py

def get_by_url(self, url: str) -> Optional[RegisteredAgent]:
    """
    Get a registered agent by URL.

    Args:
        url: URL of the agent

    Returns:
        RegisteredAgent if found, None otherwise
    """
    url = url.rstrip("/")
    name = self._url_to_name.get(url)
    return self._discovered.get(name) if name else None

get_by_skill ¶

get_by_skill(skill_id: str, *, include_unhealthy: bool = False, match_name: bool = True) -> List[RegisteredAgent]

Find agents that have a specific skill.

Searches by skill ID and optionally by skill name.

PARAMETER	DESCRIPTION
`skill_id`	Skill ID or name to search for TYPE: `str`
`include_unhealthy`	If True, include unhealthy agents TYPE: `bool` DEFAULT: `False`
`match_name`	If True, also match against skill name TYPE: `bool` DEFAULT: `True`

RETURNS	DESCRIPTION
`List[RegisteredAgent]`	List of matching RegisteredAgent instances

Source code in packages/ai-parrot/src/parrot/a2a/mesh.py

def get_by_skill(
    self,
    skill_id: str,
    *,
    include_unhealthy: bool = False,
    match_name: bool = True,
) -> List[RegisteredAgent]:
    """
    Find agents that have a specific skill.

    Searches by skill ID and optionally by skill name.

    Args:
        skill_id: Skill ID or name to search for
        include_unhealthy: If True, include unhealthy agents
        match_name: If True, also match against skill name

    Returns:
        List of matching RegisteredAgent instances
    """
    results: List[RegisteredAgent] = []
    skill_lower = skill_id.lower()

    for agent in self._discovered.values():
        # Skip unhealthy unless requested
        if not include_unhealthy and not agent.healthy:
            continue

        # Check each skill
        for skill in agent.card.skills:
            matched = skill.id.lower() == skill_lower

            if not matched and match_name:
                matched = skill_lower in skill.name.lower()

            if matched:
                results.append(agent)
                break  # Don't add same agent twice

    return results

get_by_tag ¶

get_by_tag(tag: str, *, include_unhealthy: bool = False, check_skill_tags: bool = True) -> List[RegisteredAgent]

Find agents that have a specific tag.

Searches agent-level tags and optionally skill-level tags.

PARAMETER	DESCRIPTION
`tag`	Tag to search for (case-insensitive) TYPE: `str`
`include_unhealthy`	If True, include unhealthy agents TYPE: `bool` DEFAULT: `False`
`check_skill_tags`	If True, also check tags on individual skills TYPE: `bool` DEFAULT: `True`

RETURNS	DESCRIPTION
`List[RegisteredAgent]`	List of matching RegisteredAgent instances

Source code in packages/ai-parrot/src/parrot/a2a/mesh.py

def get_by_tag(
    self,
    tag: str,
    *,
    include_unhealthy: bool = False,
    check_skill_tags: bool = True,
) -> List[RegisteredAgent]:
    """
    Find agents that have a specific tag.

    Searches agent-level tags and optionally skill-level tags.

    Args:
        tag: Tag to search for (case-insensitive)
        include_unhealthy: If True, include unhealthy agents
        check_skill_tags: If True, also check tags on individual skills

    Returns:
        List of matching RegisteredAgent instances
    """
    results: List[RegisteredAgent] = []
    tag_lower = tag.lower()

    for agent in self._discovered.values():
        # Skip unhealthy unless requested
        if not include_unhealthy and not agent.healthy:
            continue

        # Check agent-level tags
        if any(t.lower() == tag_lower for t in agent.card.tags):
            results.append(agent)
            continue

        # Check skill-level tags
        if check_skill_tags:
            for skill in agent.card.skills:
                if any(t.lower() == tag_lower for t in skill.tags):
                    results.append(agent)
                    break

    return results

search ¶

search(query: str, *, include_unhealthy: bool = False, search_fields: Optional[List[str]] = None) -> List[RegisteredAgent]

Full-text search across agent metadata.

Searches agent name, description, skill names, and skill descriptions.

PARAMETER	DESCRIPTION
`query`	Search query (case-insensitive) TYPE: `str`
`include_unhealthy`	If True, include unhealthy agents TYPE: `bool` DEFAULT: `False`
`search_fields`	Fields to search (default: name, description, skills) TYPE: `Optional[List[str]]` DEFAULT: `None`

RETURNS	DESCRIPTION
`List[RegisteredAgent]`	List of matching RegisteredAgent instances

Source code in packages/ai-parrot/src/parrot/a2a/mesh.py

def search(
    self,
    query: str,
    *,
    include_unhealthy: bool = False,
    search_fields: Optional[List[str]] = None,
) -> List[RegisteredAgent]:
    """
    Full-text search across agent metadata.

    Searches agent name, description, skill names, and skill descriptions.

    Args:
        query: Search query (case-insensitive)
        include_unhealthy: If True, include unhealthy agents
        search_fields: Fields to search (default: name, description, skills)

    Returns:
        List of matching RegisteredAgent instances
    """
    results: List[RegisteredAgent] = []
    query_lower = query.lower()

    if search_fields is None:
        search_fields = ["name", "description", "skills"]

    for agent in self._discovered.values():
        if not include_unhealthy and not agent.healthy:
            continue

        matched = False
        card = agent.card

        # Search name
        if "name" in search_fields and query_lower in card.name.lower():
            matched = True

        # Search description
        if not matched and "description" in search_fields:
            if query_lower in card.description.lower():
                matched = True

        # Search skills
        if not matched and "skills" in search_fields:
            for skill in card.skills:
                if query_lower in skill.name.lower():
                    matched = True
                    break
                if query_lower in skill.description.lower():
                    matched = True
                    break

        # Search tags
        if not matched and "tags" in search_fields:
            if any(query_lower in t.lower() for t in card.tags):
                matched = True

        if matched:
            results.append(agent)

    return results

list_healthy ¶

list_healthy() -> List[RegisteredAgent]

Get all healthy agents.

RETURNS	DESCRIPTION
`List[RegisteredAgent]`	List of healthy RegisteredAgent instances

Source code in packages/ai-parrot/src/parrot/a2a/mesh.py

def list_healthy(self) -> List[RegisteredAgent]:
    """
    Get all healthy agents.

    Returns:
        List of healthy RegisteredAgent instances
    """
    return [a for a in self._discovered.values() if a.healthy]

list_unhealthy ¶

list_unhealthy() -> List[RegisteredAgent]

Get all unhealthy agents.

RETURNS	DESCRIPTION
`List[RegisteredAgent]`	List of unhealthy RegisteredAgent instances

Source code in packages/ai-parrot/src/parrot/a2a/mesh.py

def list_unhealthy(self) -> List[RegisteredAgent]:
    """
    Get all unhealthy agents.

    Returns:
        List of unhealthy RegisteredAgent instances
    """
    return [a for a in self._discovered.values() if not a.healthy]

list_all ¶

list_all() -> List[RegisteredAgent]

Get all registered agents regardless of health status.

RETURNS	DESCRIPTION
`List[RegisteredAgent]`	List of all RegisteredAgent instances

Source code in packages/ai-parrot/src/parrot/a2a/mesh.py

def list_all(self) -> List[RegisteredAgent]:
    """
    Get all registered agents regardless of health status.

    Returns:
        List of all RegisteredAgent instances
    """
    return list(self._discovered.values())

list_by_priority ¶

list_by_priority(descending: bool = True) -> List[RegisteredAgent]

Get agents sorted by priority.

Priority comes from the endpoint configuration.

PARAMETER	DESCRIPTION
`descending`	If True, highest priority first TYPE: `bool` DEFAULT: `True`

RETURNS	DESCRIPTION
`List[RegisteredAgent]`	List of RegisteredAgent instances sorted by priority

Source code in packages/ai-parrot/src/parrot/a2a/mesh.py

def list_by_priority(self, descending: bool = True) -> List[RegisteredAgent]:
    """
    Get agents sorted by priority.

    Priority comes from the endpoint configuration.

    Args:
        descending: If True, highest priority first

    Returns:
        List of RegisteredAgent instances sorted by priority
    """
    def get_priority(agent: RegisteredAgent) -> int:
        endpoint = self._endpoints.get(agent.url)
        return endpoint.priority if endpoint else 0

    return sorted(
        self._discovered.values(),
        key=get_priority,
        reverse=descending
    )

check_health_now `async` ¶

check_health_now(name: Optional[str] = None) -> Dict[str, bool]

Trigger immediate health check.

PARAMETER	DESCRIPTION
`name`	If provided, only check this agent. Otherwise check all. TYPE: `Optional[str]` DEFAULT: `None`

RETURNS	DESCRIPTION
`Dict[str, bool]`	Dict mapping agent name to health status

Source code in packages/ai-parrot/src/parrot/a2a/mesh.py

async def check_health_now(self, name: Optional[str] = None) -> Dict[str, bool]:
    """
    Trigger immediate health check.

    Args:
        name: If provided, only check this agent. Otherwise check all.

    Returns:
        Dict mapping agent name to health status
    """
    results: Dict[str, bool] = {}

    if name:
        if agent := self._discovered.get(name):
            await self._check_agent_health(name, agent)
            results[name] = agent.healthy
    else:
        await self._perform_health_checks()
        for agent_name, agent in self._discovered.items():
            results[agent_name] = agent.healthy

    return results

on_agent_healthy ¶

on_agent_healthy(callback: AgentEventCallback) -> 'A2AMeshDiscovery'

Register callback for when an agent becomes healthy.

PARAMETER	DESCRIPTION
`callback`	Async callback(agent, event_type) TYPE: `AgentEventCallback`

RETURNS	DESCRIPTION
`'A2AMeshDiscovery'`	Self for method chaining

Source code in packages/ai-parrot/src/parrot/a2a/mesh.py

def on_agent_healthy(self, callback: AgentEventCallback) -> "A2AMeshDiscovery":
    """
    Register callback for when an agent becomes healthy.

    Args:
        callback: Async callback(agent, event_type)

    Returns:
        Self for method chaining
    """
    self._on_agent_healthy.append(callback)
    return self

on_agent_unhealthy ¶

on_agent_unhealthy(callback: AgentEventCallback) -> 'A2AMeshDiscovery'

Register callback for when an agent becomes unhealthy.

PARAMETER	DESCRIPTION
`callback`	Async callback(agent, event_type) TYPE: `AgentEventCallback`

RETURNS	DESCRIPTION
`'A2AMeshDiscovery'`	Self for method chaining

Source code in packages/ai-parrot/src/parrot/a2a/mesh.py

def on_agent_unhealthy(self, callback: AgentEventCallback) -> "A2AMeshDiscovery":
    """
    Register callback for when an agent becomes unhealthy.

    Args:
        callback: Async callback(agent, event_type)

    Returns:
        Self for method chaining
    """
    self._on_agent_unhealthy.append(callback)
    return self

on_agent_registered ¶

on_agent_registered(callback: AgentEventCallback) -> 'A2AMeshDiscovery'

Register callback for when a new agent is registered.

PARAMETER	DESCRIPTION
`callback`	Async callback(agent, event_type) TYPE: `AgentEventCallback`

RETURNS	DESCRIPTION
`'A2AMeshDiscovery'`	Self for method chaining

Source code in packages/ai-parrot/src/parrot/a2a/mesh.py

def on_agent_registered(self, callback: AgentEventCallback) -> "A2AMeshDiscovery":
    """
    Register callback for when a new agent is registered.

    Args:
        callback: Async callback(agent, event_type)

    Returns:
        Self for method chaining
    """
    self._on_agent_registered.append(callback)
    return self

on_agent_removed ¶

on_agent_removed(callback: AgentEventCallback) -> 'A2AMeshDiscovery'

Register callback for when an agent is removed.

PARAMETER	DESCRIPTION
`callback`	Async callback(agent, event_type) TYPE: `AgentEventCallback`

RETURNS	DESCRIPTION
`'A2AMeshDiscovery'`	Self for method chaining

Source code in packages/ai-parrot/src/parrot/a2a/mesh.py

def on_agent_removed(self, callback: AgentEventCallback) -> "A2AMeshDiscovery":
    """
    Register callback for when an agent is removed.

    Args:
        callback: Async callback(agent, event_type)

    Returns:
        Self for method chaining
    """
    self._on_agent_removed.append(callback)
    return self

get_info ¶

get_info() -> Dict[str, Any]

Get detailed information about the mesh state.

RETURNS	DESCRIPTION
`Dict[str, Any]`	Dictionary with mesh status information

Source code in packages/ai-parrot/src/parrot/a2a/mesh.py

def get_info(self) -> Dict[str, Any]:
    """
    Get detailed information about the mesh state.

    Returns:
        Dictionary with mesh status information
    """
    return {
        "running": self._running,
        "endpoints_configured": len(self._endpoints),
        "agents_discovered": len(self._discovered),
        "stats": self._stats.to_dict(),
        "agents": {
            name: {
                "url": agent.url,
                "healthy": agent.healthy,
                "last_seen": agent.last_seen.isoformat(),
                "skills_count": len(agent.card.skills),
                "tags": agent.card.tags,
                "failure_count": self._failure_counts.get(name, 0),
            }
            for name, agent in self._discovered.items()
        },
    }

A2AEndpoint `dataclass` ¶

A2AEndpoint(url: str, name: Optional[str] = None, auth_token: Optional[str] = None, api_key: Optional[str] = None, headers: Optional[Dict[str, str]] = None, tags: Set[str] = set(), timeout: float = 30.0, health_check_strategy: HealthCheckStrategy = HealthCheckStrategy.DISCOVERY, health_check_endpoint: Optional[str] = None, enabled: bool = True, priority: int = 0, metadata: Dict[str, Any] = dict())

Configuration for an A2A endpoint before discovery.

Represents a known endpoint that can be registered with the mesh. The actual AgentCard is fetched during discovery.

ATTRIBUTE	DESCRIPTION
`url`	Base URL of the A2A agent TYPE: `str`
`name`	Optional name hint (actual name comes from AgentCard) TYPE: `Optional[str]`
`auth_token`	Bearer token for authentication TYPE: `Optional[str]`
`api_key`	API key for X-API-Key header TYPE: `Optional[str]`
`headers`	Additional HTTP headers TYPE: `Optional[Dict[str, str]]`
`tags`	Local tags for categorization (merged with agent's tags) TYPE: `Set[str]`
`timeout`	Request timeout for this endpoint TYPE: `float`
`health_check_strategy`	How to check health for this endpoint TYPE: `HealthCheckStrategy`
`health_check_endpoint`	Custom health check endpoint (if strategy is CUSTOM) TYPE: `Optional[str]`
`enabled`	Whether this endpoint is enabled TYPE: `bool`
`priority`	Priority for load balancing (higher = preferred) TYPE: `int`
`metadata`	Additional metadata TYPE: `Dict[str, Any]`

HealthCheckStrategy ¶

Bases: str, Enum

Strategy for health checking agents.

AgentStatus ¶

Bases: str, Enum

Status of an agent in the mesh.

DiscoveryStats `dataclass` ¶

DiscoveryStats(total_registered: int = 0, total_healthy: int = 0, total_unhealthy: int = 0, last_health_check: Optional[datetime] = None, health_checks_performed: int = 0, discovery_errors: int = 0)

Statistics about mesh discovery operations.

to_dict ¶

to_dict() -> Dict[str, Any]

Convert stats to dictionary.

Source code in packages/ai-parrot/src/parrot/a2a/mesh.py

def to_dict(self) -> Dict[str, Any]:
    """Convert stats to dictionary."""
    return {
        "total_registered": self.total_registered,
        "total_healthy": self.total_healthy,
        "total_unhealthy": self.total_unhealthy,
        "last_health_check": self.last_health_check.isoformat() if self.last_health_check else None,
        "health_checks_performed": self.health_checks_performed,
        "discovery_errors": self.discovery_errors,
    }

A2AProxyRouter ¶

A2AProxyRouter(mesh: A2AMeshDiscovery, *, name: str = 'A2ARouter', description: str = 'A2A Proxy Router', version: str = '1.0.0', aggregate_skills: bool = True, skill_prefix_with_agent: bool = True, default_timeout: float = 60.0, base_path: str = '/a2a', tags: Optional[List[str]] = None, capabilities: Optional[AgentCapabilities] = None)

Proxy/Gateway for routing requests to A2A agents without LLM processing.

This router receives requests and forwards them to appropriate downstream agents based on configurable routing rules. No LLM is involved in the routing decision - it's pure rule-based matching.

The router can also expose itself as an A2A-compliant server, presenting an aggregated view of all downstream agents' capabilities.

Use Cases

API Gateway: Single entry point for multiple specialized agents
Load Balancer: Distribute requests across equivalent agents
Router: Direct requests based on content/intent
Facade: Hide internal agent topology from clients

Example

Setup¶

mesh = A2AMeshDiscovery() await mesh.register("http://agent1:8080") await mesh.register("http://agent2:8080")

router = A2AProxyRouter(mesh, name="Gateway")

Add routing rules¶

router.route_by_skill("data_analysis", "DataAnalyst") router.route_by_tag("customer", "SupportBot") router.route_by_regex(r"urgent|emergency", "PriorityHandler") router.set_default("GeneralAssistant")

Route a message (no LLM involved!)¶

task = await router.route_message("Analyze this data...") print(task.artifacts[0].parts[0].text)

Expose as A2A server¶

app = web.Application() router.setup(app)

Initialize the A2A Proxy Router.

PARAMETER	DESCRIPTION
`mesh`	A2AMeshDiscovery instance for agent lookup TYPE: `A2AMeshDiscovery`
`name`	Name for this router (used in AgentCard) TYPE: `str` DEFAULT: `'A2ARouter'`
`description`	Description for this router TYPE: `str` DEFAULT: `'A2A Proxy Router'`
`version`	Version string TYPE: `str` DEFAULT: `'1.0.0'`
`aggregate_skills`	If True, aggregate skills from all downstream agents TYPE: `bool` DEFAULT: `True`
`skill_prefix_with_agent`	If True, prefix aggregated skills with agent name TYPE: `bool` DEFAULT: `True`
`default_timeout`	Default timeout for proxied requests TYPE: `float` DEFAULT: `60.0`
`base_path`	URL prefix for A2A endpoints when exposed as server TYPE: `str` DEFAULT: `'/a2a'`
`tags`	Tags for the router's AgentCard TYPE: `Optional[List[str]]` DEFAULT: `None`
`capabilities`	Capabilities to advertise TYPE: `Optional[AgentCapabilities]` DEFAULT: `None`

Source code in packages/ai-parrot/src/parrot/a2a/router.py

def __init__(
    self,
    mesh: A2AMeshDiscovery,
    *,
    name: str = "A2ARouter",
    description: str = "A2A Proxy Router",
    version: str = "1.0.0",
    aggregate_skills: bool = True,
    skill_prefix_with_agent: bool = True,
    default_timeout: float = 60.0,
    base_path: str = "/a2a",
    tags: Optional[List[str]] = None,
    capabilities: Optional[AgentCapabilities] = None,
):
    """
    Initialize the A2A Proxy Router.

    Args:
        mesh: A2AMeshDiscovery instance for agent lookup
        name: Name for this router (used in AgentCard)
        description: Description for this router
        version: Version string
        aggregate_skills: If True, aggregate skills from all downstream agents
        skill_prefix_with_agent: If True, prefix aggregated skills with agent name
        default_timeout: Default timeout for proxied requests
        base_path: URL prefix for A2A endpoints when exposed as server
        tags: Tags for the router's AgentCard
        capabilities: Capabilities to advertise
    """
    self.mesh = mesh
    self.name = name
    self.description = description
    self.version = version
    self.aggregate_skills = aggregate_skills
    self.skill_prefix_with_agent = skill_prefix_with_agent
    self.default_timeout = default_timeout
    self.base_path = base_path.rstrip("/")
    self.tags = tags or ["gateway", "router", "proxy"]
    self.capabilities = capabilities or AgentCapabilities(streaming=True)

    # Routing rules
    self._rules: List[RoutingRule] = []
    self._default_agent: Optional[str] = None

    # Runtime state
    self._stats = ProxyStats()
    self._app: Optional[web.Application] = None
    self._agent_card_cache: Optional[AgentCard] = None
    self._agent_card_cache_time: Optional[datetime] = None
    self._cache_ttl_seconds: float = 60.0  # Refresh aggregated card every 60s

    # Active client connections (reused for performance)
    self._client_cache: Dict[str, A2AClient] = {}

    self.logger = logging.getLogger(f"A2A.Router.{name}")

stats `property` ¶

stats: ProxyStats

Get current statistics.

add_route ¶

add_route(pattern: str, target: Union[str, List[str]], *, strategy: RoutingStrategy = RoutingStrategy.SKILL_MATCH, priority: int = 0, load_balance: LoadBalanceStrategy = LoadBalanceStrategy.FIRST_HEALTHY, weights: Optional[Dict[str, float]] = None, transform_request: Optional[TransformFunc] = None, transform_response: Optional[ResponseTransformFunc] = None, enabled: bool = True, metadata: Optional[Dict[str, Any]] = None) -> 'A2AProxyRouter'

Add a routing rule.

PARAMETER	DESCRIPTION
`pattern`	Pattern to match (interpretation depends on strategy) TYPE: `str`
`target`	Agent name or list of agent names TYPE: `Union[str, List[str]]`
`strategy`	How to match the pattern TYPE: `RoutingStrategy` DEFAULT: `SKILL_MATCH`
`priority`	Rule priority (higher = evaluated first) TYPE: `int` DEFAULT: `0`
`load_balance`	How to select among multiple targets TYPE: `LoadBalanceStrategy` DEFAULT: `FIRST_HEALTHY`
`weights`	Weights for weighted load balancing TYPE: `Optional[Dict[str, float]]` DEFAULT: `None`
`transform_request`	Async function to transform request TYPE: `Optional[TransformFunc]` DEFAULT: `None`
`transform_response`	Async function to transform response TYPE: `Optional[ResponseTransformFunc]` DEFAULT: `None`
`enabled`	Whether rule is active TYPE: `bool` DEFAULT: `True`
`metadata`	Additional metadata TYPE: `Optional[Dict[str, Any]]` DEFAULT: `None`

RETURNS	DESCRIPTION
`'A2AProxyRouter'`	Self for method chaining

Example

router.add_route( "data_analysis", ["Analyst1", "Analyst2"], strategy=RoutingStrategy.SKILL_MATCH, load_balance=LoadBalanceStrategy.ROUND_ROBIN, priority=10 )

Source code in packages/ai-parrot/src/parrot/a2a/router.py

def add_route(
    self,
    pattern: str,
    target: Union[str, List[str]],
    *,
    strategy: RoutingStrategy = RoutingStrategy.SKILL_MATCH,
    priority: int = 0,
    load_balance: LoadBalanceStrategy = LoadBalanceStrategy.FIRST_HEALTHY,
    weights: Optional[Dict[str, float]] = None,
    transform_request: Optional[TransformFunc] = None,
    transform_response: Optional[ResponseTransformFunc] = None,
    enabled: bool = True,
    metadata: Optional[Dict[str, Any]] = None,
) -> "A2AProxyRouter":
    """
    Add a routing rule.

    Args:
        pattern: Pattern to match (interpretation depends on strategy)
        target: Agent name or list of agent names
        strategy: How to match the pattern
        priority: Rule priority (higher = evaluated first)
        load_balance: How to select among multiple targets
        weights: Weights for weighted load balancing
        transform_request: Async function to transform request
        transform_response: Async function to transform response
        enabled: Whether rule is active
        metadata: Additional metadata

    Returns:
        Self for method chaining

    Example:
        router.add_route(
            "data_analysis",
            ["Analyst1", "Analyst2"],
            strategy=RoutingStrategy.SKILL_MATCH,
            load_balance=LoadBalanceStrategy.ROUND_ROBIN,
            priority=10
        )
    """
    targets = [target] if isinstance(target, str) else list(target)

    rule = RoutingRule(
        pattern=pattern,
        strategy=strategy,
        target_agents=targets,
        priority=priority,
        load_balance=load_balance,
        weights=weights,
        transform_request=transform_request,
        transform_response=transform_response,
        enabled=enabled,
        metadata=metadata or {},
    )

    self._rules.append(rule)
    self._sort_rules()

    self.logger.debug(
        f"Added route: {strategy.value}:{pattern} -> {targets}"
    )

    return self

route_by_skill ¶

route_by_skill(skill_id: str, target: Union[str, List[str]], **kwargs) -> 'A2AProxyRouter'

Add routing rule that matches by skill ID.

PARAMETER	DESCRIPTION
`skill_id`	Skill ID to match TYPE: `str`
`target`	Target agent(s) TYPE: `Union[str, List[str]]`
`**kwargs`	Additional RoutingRule parameters DEFAULT: `{}`

RETURNS	DESCRIPTION
`'A2AProxyRouter'`	Self for method chaining

Source code in packages/ai-parrot/src/parrot/a2a/router.py

def route_by_skill(
    self,
    skill_id: str,
    target: Union[str, List[str]],
    **kwargs
) -> "A2AProxyRouter":
    """
    Add routing rule that matches by skill ID.

    Args:
        skill_id: Skill ID to match
        target: Target agent(s)
        **kwargs: Additional RoutingRule parameters

    Returns:
        Self for method chaining
    """
    return self.add_route(
        skill_id,
        target,
        strategy=RoutingStrategy.SKILL_MATCH,
        **kwargs
    )

route_by_skill_name ¶

route_by_skill_name(skill_name: str, target: Union[str, List[str]], **kwargs) -> 'A2AProxyRouter'

Add routing rule that matches by skill name (partial match).

PARAMETER	DESCRIPTION
`skill_name`	Skill name pattern to match TYPE: `str`
`target`	Target agent(s) TYPE: `Union[str, List[str]]`
`**kwargs`	Additional RoutingRule parameters DEFAULT: `{}`

RETURNS	DESCRIPTION
`'A2AProxyRouter'`	Self for method chaining

Source code in packages/ai-parrot/src/parrot/a2a/router.py

def route_by_skill_name(
    self,
    skill_name: str,
    target: Union[str, List[str]],
    **kwargs
) -> "A2AProxyRouter":
    """
    Add routing rule that matches by skill name (partial match).

    Args:
        skill_name: Skill name pattern to match
        target: Target agent(s)
        **kwargs: Additional RoutingRule parameters

    Returns:
        Self for method chaining
    """
    return self.add_route(
        skill_name,
        target,
        strategy=RoutingStrategy.SKILL_NAME,
        **kwargs
    )

route_by_tag ¶

route_by_tag(tag: str, target: Union[str, List[str]], **kwargs) -> 'A2AProxyRouter'

Add routing rule that matches by tag.

PARAMETER	DESCRIPTION
`tag`	Tag to match TYPE: `str`
`target`	Target agent(s) TYPE: `Union[str, List[str]]`
`**kwargs`	Additional RoutingRule parameters DEFAULT: `{}`

RETURNS	DESCRIPTION
`'A2AProxyRouter'`	Self for method chaining

Source code in packages/ai-parrot/src/parrot/a2a/router.py

def route_by_tag(
    self,
    tag: str,
    target: Union[str, List[str]],
    **kwargs
) -> "A2AProxyRouter":
    """
    Add routing rule that matches by tag.

    Args:
        tag: Tag to match
        target: Target agent(s)
        **kwargs: Additional RoutingRule parameters

    Returns:
        Self for method chaining
    """
    return self.add_route(
        tag,
        target,
        strategy=RoutingStrategy.TAG_MATCH,
        **kwargs
    )

route_by_regex ¶

route_by_regex(pattern: str, target: Union[str, List[str]], **kwargs) -> 'A2AProxyRouter'

Add routing rule that matches by regex pattern in the message.

PARAMETER	DESCRIPTION
`pattern`	Regex pattern to match against message content TYPE: `str`
`target`	Target agent(s) TYPE: `Union[str, List[str]]`
`**kwargs`	Additional RoutingRule parameters DEFAULT: `{}`

RETURNS	DESCRIPTION
`'A2AProxyRouter'`	Self for method chaining

Example

router.route_by_regex( r"urgent|emergency|critical", "PriorityHandler", priority=100 # High priority )

Source code in packages/ai-parrot/src/parrot/a2a/router.py

def route_by_regex(
    self,
    pattern: str,
    target: Union[str, List[str]],
    **kwargs
) -> "A2AProxyRouter":
    """
    Add routing rule that matches by regex pattern in the message.

    Args:
        pattern: Regex pattern to match against message content
        target: Target agent(s)
        **kwargs: Additional RoutingRule parameters

    Returns:
        Self for method chaining

    Example:
        router.route_by_regex(
            r"urgent|emergency|critical",
            "PriorityHandler",
            priority=100  # High priority
        )
    """
    return self.add_route(
        pattern,
        target,
        strategy=RoutingStrategy.REGEX,
        **kwargs
    )

route_round_robin ¶

route_round_robin(agents: List[str], **kwargs) -> 'A2AProxyRouter'

Add round-robin routing across multiple agents.

All requests matching this rule will be distributed evenly across the specified agents.

PARAMETER	DESCRIPTION
`agents`	List of agent names to rotate through TYPE: `List[str]`
`**kwargs`	Additional RoutingRule parameters DEFAULT: `{}`

RETURNS	DESCRIPTION
`'A2AProxyRouter'`	Self for method chaining

Source code in packages/ai-parrot/src/parrot/a2a/router.py

def route_round_robin(
    self,
    agents: List[str],
    **kwargs
) -> "A2AProxyRouter":
    """
    Add round-robin routing across multiple agents.

    All requests matching this rule will be distributed evenly
    across the specified agents.

    Args:
        agents: List of agent names to rotate through
        **kwargs: Additional RoutingRule parameters

    Returns:
        Self for method chaining
    """
    return self.add_route(
        "*",  # Match all
        agents,
        strategy=RoutingStrategy.ROUND_ROBIN,
        load_balance=LoadBalanceStrategy.ROUND_ROBIN,
        **kwargs
    )

set_default ¶

set_default(agent_name: str) -> 'A2AProxyRouter'

Set the default agent for requests that don't match any rule.

PARAMETER	DESCRIPTION
`agent_name`	Name of the default agent TYPE: `str`

RETURNS	DESCRIPTION
`'A2AProxyRouter'`	Self for method chaining

Source code in packages/ai-parrot/src/parrot/a2a/router.py

def set_default(self, agent_name: str) -> "A2AProxyRouter":
    """
    Set the default agent for requests that don't match any rule.

    Args:
        agent_name: Name of the default agent

    Returns:
        Self for method chaining
    """
    self._default_agent = agent_name
    self.logger.debug(f"Set default agent: {agent_name}")
    return self

remove_route ¶

remove_route(pattern: str, strategy: Optional[RoutingStrategy] = None) -> bool

Remove a routing rule.

PARAMETER	DESCRIPTION
`pattern`	Pattern of the rule to remove TYPE: `str`
`strategy`	If provided, only remove rule with matching strategy TYPE: `Optional[RoutingStrategy]` DEFAULT: `None`

RETURNS	DESCRIPTION
`bool`	True if rule was removed, False if not found

Source code in packages/ai-parrot/src/parrot/a2a/router.py

def remove_route(self, pattern: str, strategy: Optional[RoutingStrategy] = None) -> bool:
    """
    Remove a routing rule.

    Args:
        pattern: Pattern of the rule to remove
        strategy: If provided, only remove rule with matching strategy

    Returns:
        True if rule was removed, False if not found
    """
    for i, rule in enumerate(self._rules):
        if rule.pattern == pattern:
            if strategy is None or rule.strategy == strategy:
                del self._rules[i]
                self.logger.debug(f"Removed route: {pattern}")
                return True
    return False

clear_routes ¶

clear_routes() -> 'A2AProxyRouter'

Clear all routing rules.

RETURNS	DESCRIPTION
`'A2AProxyRouter'`	Self for method chaining

Source code in packages/ai-parrot/src/parrot/a2a/router.py

def clear_routes(self) -> "A2AProxyRouter":
    """
    Clear all routing rules.

    Returns:
        Self for method chaining
    """
    self._rules.clear()
    self._default_agent = None
    self.logger.debug("Cleared all routes")
    return self

list_routes ¶

list_routes() -> List[Dict[str, Any]]

List all configured routing rules.

RETURNS	DESCRIPTION
`List[Dict[str, Any]]`	List of rule configurations

Source code in packages/ai-parrot/src/parrot/a2a/router.py

def list_routes(self) -> List[Dict[str, Any]]:
    """
    List all configured routing rules.

    Returns:
        List of rule configurations
    """
    return [
        {
            "pattern": rule.pattern,
            "strategy": rule.strategy.value,
            "targets": rule.target_agents,
            "priority": rule.priority,
            "load_balance": rule.load_balance.value,
            "enabled": rule.enabled,
        }
        for rule in self._rules
    ]

find_target ¶

find_target(message: str, *, skill_id: Optional[str] = None, tags: Optional[List[str]] = None) -> RoutingResult

Find the target agent for a request.

Evaluates all rules in priority order and returns the first match.

PARAMETER	DESCRIPTION
`message`	Message content TYPE: `str`
`skill_id`	Optional skill ID TYPE: `Optional[str]` DEFAULT: `None`
`tags`	Optional tags TYPE: `Optional[List[str]]` DEFAULT: `None`

RETURNS	DESCRIPTION
`RoutingResult`	RoutingResult with matched rule and target agent

Source code in packages/ai-parrot/src/parrot/a2a/router.py

def find_target(
    self,
    message: str,
    *,
    skill_id: Optional[str] = None,
    tags: Optional[List[str]] = None,
) -> RoutingResult:
    """
    Find the target agent for a request.

    Evaluates all rules in priority order and returns the first match.

    Args:
        message: Message content
        skill_id: Optional skill ID
        tags: Optional tags

    Returns:
        RoutingResult with matched rule and target agent
    """
    # Try each rule in priority order
    for rule in self._rules:
        if self._match_rule(rule, message, skill_id, tags):
            if target := self._select_target(rule):
                return RoutingResult(
                    matched=True,
                    rule=rule,
                    target_agent=target,
                    match_details={
                        "pattern": rule.pattern,
                        "strategy": rule.strategy.value,
                    }
                )

    # Try default agent
    if self._default_agent:
        if agent := self.mesh.get(self._default_agent):
            if agent.healthy:
                return RoutingResult(
                    matched=True,
                    rule=None,
                    target_agent=agent,
                    match_details={"default": True}
                )

    # No match found
    return RoutingResult(matched=False)

route_message `async` ¶

route_message(message: str, *, skill_id: Optional[str] = None, tags: Optional[List[str]] = None, context_id: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None) -> Task

Route a message to the appropriate agent and return the response.

This is a PASSTHROUGH operation - no LLM is involved. The message is forwarded as-is to the matched agent.

PARAMETER	DESCRIPTION
`message`	Message content to route TYPE: `str`
`skill_id`	Optional skill ID to help with routing TYPE: `Optional[str]` DEFAULT: `None`
`tags`	Optional tags to help with routing TYPE: `Optional[List[str]]` DEFAULT: `None`
`context_id`	Optional context ID for conversations TYPE: `Optional[str]` DEFAULT: `None`
`metadata`	Optional metadata TYPE: `Optional[Dict[str, Any]]` DEFAULT: `None`
`timeout`	Optional timeout override TYPE: `Optional[float]` DEFAULT: `None`

RETURNS	DESCRIPTION
`Task`	Task with the response from the downstream agent

RAISES	DESCRIPTION
`ValueError`	If no target agent found
`ConnectionError`	If target agent is unreachable

Source code in packages/ai-parrot/src/parrot/a2a/router.py

async def route_message(
    self,
    message: str,
    *,
    skill_id: Optional[str] = None,
    tags: Optional[List[str]] = None,
    context_id: Optional[str] = None,
    metadata: Optional[Dict[str, Any]] = None,
    timeout: Optional[float] = None,
) -> Task:
    """
    Route a message to the appropriate agent and return the response.

    This is a PASSTHROUGH operation - no LLM is involved.
    The message is forwarded as-is to the matched agent.

    Args:
        message: Message content to route
        skill_id: Optional skill ID to help with routing
        tags: Optional tags to help with routing
        context_id: Optional context ID for conversations
        metadata: Optional metadata
        timeout: Optional timeout override

    Returns:
        Task with the response from the downstream agent

    Raises:
        ValueError: If no target agent found
        ConnectionError: If target agent is unreachable
    """
    start_time = time.monotonic()
    self._stats.requests_total += 1
    self._stats.last_request_time = datetime.now(timezone.utc)

    # Find target
    result = self.find_target(message, skill_id=skill_id, tags=tags)

    if not result.matched or not result.target_agent:
        self._stats.requests_no_match += 1
        raise ValueError("No target agent found for request")

    agent = result.target_agent
    rule = result.rule

    self.logger.debug(
        f"Routing to {agent.card.name}: {result.match_details}"
    )

    try:
        # Apply request transformation if configured
        transformed_message = message
        if rule and rule.transform_request:
            transformed_message = await rule.transform_request(
                message,
                {"skill_id": skill_id, "tags": tags, "metadata": metadata}
            )

        # Get or create client
        client = await self._get_client(agent)

        # Send message (PASSTHROUGH - no LLM!)
        task = await client.send_message(
            transformed_message,
            context_id=context_id,
            metadata=metadata,
        )

        # Apply response transformation if configured
        if rule and rule.transform_response:
            task = await rule.transform_response(
                task,
                {"agent": agent.card.name, "rule": rule.pattern}
            )

        # Update stats
        elapsed_ms = (time.monotonic() - start_time) * 1000
        self._stats.requests_routed += 1
        self._stats.total_latency_ms += elapsed_ms
        self._stats.agents_used[agent.card.name] = \
            self._stats.agents_used.get(agent.card.name, 0) + 1

        if rule:
            self._stats.rules_matched[rule.pattern] = \
                self._stats.rules_matched.get(rule.pattern, 0) + 1

        return task

    except Exception as e:
        self._stats.requests_failed += 1
        self.logger.error(f"Routing failed to {agent.card.name}: {e}")
        raise

route_message_stream `async` ¶

route_message_stream(message: str, *, skill_id: Optional[str] = None, tags: Optional[List[str]] = None, context_id: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None) -> AsyncIterator[str]

Route a message and stream the response.

PARAMETER	DESCRIPTION
`message`	Message to route TYPE: `str`
`skill_id`	Optional skill ID TYPE: `Optional[str]` DEFAULT: `None`
`tags`	Optional tags TYPE: `Optional[List[str]]` DEFAULT: `None`
`context_id`	Optional context ID TYPE: `Optional[str]` DEFAULT: `None`
`metadata`	Optional metadata TYPE: `Optional[Dict[str, Any]]` DEFAULT: `None`

YIELDS	DESCRIPTION
`AsyncIterator[str]`	Text chunks as they arrive from the downstream agent

Source code in packages/ai-parrot/src/parrot/a2a/router.py

async def route_message_stream(
    self,
    message: str,
    *,
    skill_id: Optional[str] = None,
    tags: Optional[List[str]] = None,
    context_id: Optional[str] = None,
    metadata: Optional[Dict[str, Any]] = None,
) -> AsyncIterator[str]:
    """
    Route a message and stream the response.

    Args:
        message: Message to route
        skill_id: Optional skill ID
        tags: Optional tags
        context_id: Optional context ID
        metadata: Optional metadata

    Yields:
        Text chunks as they arrive from the downstream agent
    """
    result = self.find_target(message, skill_id=skill_id, tags=tags)

    if not result.matched or not result.target_agent:
        raise ValueError("No target agent found for request")

    agent = result.target_agent
    rule = result.rule

    # Apply request transformation
    transformed_message = message
    if rule and rule.transform_request:
        transformed_message = await rule.transform_request(
            message,
            {"skill_id": skill_id, "tags": tags}
        )

    # Get client and stream
    client = await self._get_client(agent)

    async for chunk in client.stream_message(
        transformed_message,
        context_id=context_id,
        metadata=metadata,
    ):
        yield chunk

invoke_skill `async` ¶

invoke_skill(skill_id: str, params: Optional[Dict[str, Any]] = None, *, agent_name: Optional[str] = None, context_id: Optional[str] = None) -> Any

Invoke a specific skill on a remote agent.

If agent_name is not provided, finds an agent with the skill.

PARAMETER	DESCRIPTION
`skill_id`	ID of the skill to invoke TYPE: `str`
`params`	Parameters for the skill TYPE: `Optional[Dict[str, Any]]` DEFAULT: `None`
`agent_name`	Optional specific agent to use TYPE: `Optional[str]` DEFAULT: `None`
`context_id`	Optional context ID TYPE: `Optional[str]` DEFAULT: `None`

RETURNS	DESCRIPTION
`Any`	Skill result from the downstream agent

RAISES	DESCRIPTION
`ValueError`	If no agent found with the skill

Source code in packages/ai-parrot/src/parrot/a2a/router.py

async def invoke_skill(
    self,
    skill_id: str,
    params: Optional[Dict[str, Any]] = None,
    *,
    agent_name: Optional[str] = None,
    context_id: Optional[str] = None,
) -> Any:
    """
    Invoke a specific skill on a remote agent.

    If agent_name is not provided, finds an agent with the skill.

    Args:
        skill_id: ID of the skill to invoke
        params: Parameters for the skill
        agent_name: Optional specific agent to use
        context_id: Optional context ID

    Returns:
        Skill result from the downstream agent

    Raises:
        ValueError: If no agent found with the skill
    """
    self._stats.requests_total += 1

    # Find agent with skill
    if agent_name:
        agent = self.mesh.get(agent_name)
        if not agent or not agent.healthy:
            raise ValueError(f"Agent {agent_name} not available")
    else:
        # Search for agent with this skill
        agents = self.mesh.get_by_skill(skill_id)
        if not agents:
            raise ValueError(f"No agent found with skill: {skill_id}")
        agent = agents[0]

    try:
        client = await self._get_client(agent)
        result = await client.invoke_skill(skill_id, params, context_id=context_id)

        self._stats.requests_routed += 1
        self._stats.agents_used[agent.card.name] = \
            self._stats.agents_used.get(agent.card.name, 0) + 1

        return result

    except Exception:
        self._stats.requests_failed += 1
        raise

close_clients `async` ¶

close_clients() -> None

Close all cached client connections.

Source code in packages/ai-parrot/src/parrot/a2a/router.py

async def close_clients(self) -> None:
    """Close all cached client connections."""
    for client in self._client_cache.values():
        with contextlib.suppress(Exception):
            await client.disconnect()
    self._client_cache.clear()

ask `async` ¶

ask(message: str, *, agent: Optional[str] = None, **kwargs) -> str

Shortcut: send message and get response as string.

PARAMETER	DESCRIPTION
`message`	Message to send TYPE: `str`
`agent`	Optional specific agent to use TYPE: `Optional[str]` DEFAULT: `None`
`**kwargs`	Additional arguments for route_message DEFAULT: `{}`

RETURNS	DESCRIPTION
`str`	Response text

Source code in packages/ai-parrot/src/parrot/a2a/router.py

async def ask(
    self,
    message: str,
    *,
    agent: Optional[str] = None,
    **kwargs
) -> str:
    """
    Shortcut: send message and get response as string.

    Args:
        message: Message to send
        agent: Optional specific agent to use
        **kwargs: Additional arguments for route_message

    Returns:
        Response text
    """
    if agent:
        # Direct to specific agent
        target = self.mesh.get(agent)
        if not target:
            raise ValueError(f"Agent {agent} not found")

        client = await self._get_client(target)
        task = await client.send_message(message, **kwargs)
    else:
        # Route based on rules
        task = await self.route_message(message, **kwargs)

    if task.status.state == TaskState.FAILED:
        error = task.status.message.get_text() if task.status.message else "Unknown error"
        raise RuntimeError(f"Agent error: {error}")

    if task.artifacts and task.artifacts[0].parts:
        return task.artifacts[0].parts[0].text or ""

    return ""

fan_out `async` ¶

fan_out(message: str, agents: List[str], *, timeout: float = 60.0, **kwargs) -> Dict[str, Union[str, Exception]]

Send message to multiple agents in parallel.

PARAMETER	DESCRIPTION
`message`	Message to send TYPE: `str`
`agents`	List of agent names TYPE: `List[str]`
`timeout`	Timeout for each request TYPE: `float` DEFAULT: `60.0`
`**kwargs`	Additional arguments DEFAULT: `{}`

RETURNS	DESCRIPTION
`Dict[str, Union[str, Exception]]`	Dict mapping agent name to response or exception

Source code in packages/ai-parrot/src/parrot/a2a/router.py

async def fan_out(
    self,
    message: str,
    agents: List[str],
    *,
    timeout: float = 60.0,
    **kwargs
) -> Dict[str, Union[str, Exception]]:
    """
    Send message to multiple agents in parallel.

    Args:
        message: Message to send
        agents: List of agent names
        timeout: Timeout for each request
        **kwargs: Additional arguments

    Returns:
        Dict mapping agent name to response or exception
    """
    async def call_agent(name: str) -> Tuple[str, Union[str, Exception]]:
        try:
            result = await self.ask(message, agent=name, timeout=timeout, **kwargs)
            return name, result
        except Exception as e:
            return name, e

    results = await asyncio.gather(
        *[call_agent(name) for name in agents],
        return_exceptions=False
    )

    return dict(results)

pipeline `async` ¶

pipeline(message: str, agents: List[str], **kwargs) -> str

Execute a sequential pipeline of agents.

Each agent's output becomes the next agent's input.

PARAMETER	DESCRIPTION
`message`	Initial message TYPE: `str`
`agents`	Ordered list of agent names TYPE: `List[str]`
`**kwargs`	Additional arguments DEFAULT: `{}`

RETURNS	DESCRIPTION
`str`	Final output from the last agent

Source code in packages/ai-parrot/src/parrot/a2a/router.py

async def pipeline(
    self,
    message: str,
    agents: List[str],
    **kwargs
) -> str:
    """
    Execute a sequential pipeline of agents.

    Each agent's output becomes the next agent's input.

    Args:
        message: Initial message
        agents: Ordered list of agent names
        **kwargs: Additional arguments

    Returns:
        Final output from the last agent
    """
    current_input = message

    for name in agents:
        current_input = await self.ask(current_input, agent=name, **kwargs)

    return current_input

get_agent_card ¶

get_agent_card(force_refresh: bool = False) -> AgentCard

Get the AgentCard for this router.

When aggregate_skills is True, the card includes skills from all downstream agents, making this router appear as a single agent with all capabilities.

PARAMETER	DESCRIPTION
`force_refresh`	Force refresh of cached card TYPE: `bool` DEFAULT: `False`

RETURNS	DESCRIPTION
`AgentCard`	AgentCard representing this router

Source code in packages/ai-parrot/src/parrot/a2a/router.py

def get_agent_card(self, force_refresh: bool = False) -> AgentCard:
    """
    Get the AgentCard for this router.

    When aggregate_skills is True, the card includes skills from all
    downstream agents, making this router appear as a single agent
    with all capabilities.

    Args:
        force_refresh: Force refresh of cached card

    Returns:
        AgentCard representing this router
    """
    now = datetime.now(timezone.utc)

    # Check cache
    if (not force_refresh and self._agent_card_cache and self._agent_card_cache_time and (now - self._agent_card_cache_time).total_seconds() < self._cache_ttl_seconds):  # noqa
        return self._agent_card_cache

    # Build skills list
    skills: List[AgentSkill] = []
    all_tags: Set[str] = set(self.tags)

    if self.aggregate_skills:
        for agent in self.mesh.list_healthy():
            for skill in agent.card.skills:
                # Create aggregated skill
                if self.skill_prefix_with_agent:
                    skill_id = f"{agent.card.name.lower().replace(' ', '_')}:{skill.id}"
                    skill_name = f"[{agent.card.name}] {skill.name}"
                else:
                    skill_id = skill.id
                    skill_name = skill.name

                aggregated = AgentSkill(
                    id=skill_id,
                    name=skill_name,
                    description=skill.description,
                    tags=skill.tags + [agent.card.name.lower()],
                    input_schema=skill.input_schema,
                    examples=skill.examples,
                )
                skills.append(aggregated)
                all_tags.update(skill.tags)

    # Add a generic "route" skill
    skills.append(AgentSkill(
        id="route",
        name="Route Message",
        description="Route a message to the appropriate downstream agent",
        tags=["routing", "gateway"],
    ))

    card = AgentCard(
        name=self.name,
        description=self.description,
        version=self.version,
        # supported_interfaces is populated with the mount URL in
        # _handle_discovery (the flat v0.3 `url` is now a derived property).
        supported_interfaces=[],
        skills=skills,
        capabilities=self.capabilities,
        tags=list(all_tags),
    )

    # Cache
    self._agent_card_cache = card
    self._agent_card_cache_time = now

    return card

setup ¶

setup(app: Application, base_path: Optional[str] = None) -> None

Mount the router as an A2A server on an aiohttp application.

This allows external services to discover and use this router as if it were a regular A2A agent.

PARAMETER	DESCRIPTION
`app`	aiohttp Application TYPE: `Application`
`base_path`	Optional path prefix override TYPE: `Optional[str]` DEFAULT: `None`

Source code in packages/ai-parrot/src/parrot/a2a/router.py

def setup(
    self,
    app: web.Application,
    base_path: Optional[str] = None,
) -> None:
    """
    Mount the router as an A2A server on an aiohttp application.

    This allows external services to discover and use this router
    as if it were a regular A2A agent.

    Args:
        app: aiohttp Application
        base_path: Optional path prefix override
    """
    path = base_path or self.base_path

    # Discovery endpoints (v1.0 agent-card.json + v0.3 agent.json).
    app.router.add_get(
        "/.well-known/agent-card.json",
        self._handle_discovery
    )
    app.router.add_get(
        "/.well-known/agent.json",
        self._handle_discovery
    )

    # A2A message endpoints
    app.router.add_post(
        f"{path}/message/send",
        self._handle_message
    )
    app.router.add_post(
        f"{path}/message/stream",
        self._handle_stream
    )

    # Stats endpoint
    app.router.add_get(
        f"{path}/stats",
        self._handle_stats
    )

    # Routes endpoint
    app.router.add_get(
        f"{path}/routes",
        self._handle_routes
    )

    # Cleanup on shutdown
    app.on_cleanup.append(self._cleanup)

    self._app = app
    self.logger.info(f"A2A Router mounted at {path}")

get_info ¶

get_info() -> Dict[str, Any]

Get detailed information about the router state.

Source code in packages/ai-parrot/src/parrot/a2a/router.py

def get_info(self) -> Dict[str, Any]:
    """Get detailed information about the router state."""
    return {
        "name": self.name,
        "description": self.description,
        "version": self.version,
        "rules_count": len(self._rules),
        "default_agent": self._default_agent,
        "mesh_agents": len(self.mesh),
        "mesh_healthy": len(self.mesh.list_healthy()),
        "cached_clients": len(self._client_cache),
        "stats": self._stats.to_dict(),
    }

RoutingStrategy ¶

Bases: str, Enum

Strategy for selecting target agent.

LoadBalanceStrategy ¶

Bases: str, Enum

Strategy for load balancing across multiple target agents.

RoutingRule `dataclass` ¶

RoutingRule(pattern: str, strategy: RoutingStrategy, target_agents: List[str] = list(), priority: int = 0, load_balance: LoadBalanceStrategy = LoadBalanceStrategy.FIRST_HEALTHY, weights: Optional[Dict[str, float]] = None, transform_request: Optional[TransformFunc] = None, transform_response: Optional[ResponseTransformFunc] = None, enabled: bool = True, metadata: Dict[str, Any] = dict(), _compiled_regex: Optional[Pattern] = None, _round_robin_counter: int = 0, _last_used: Dict[str, datetime] = dict())

Defines a routing rule for matching requests to agents.

ATTRIBUTE	DESCRIPTION
`pattern`	Pattern to match (skill_id, tag, regex, or "" for default) TYPE:* `str`
`strategy`	How to match the pattern against requests TYPE: `RoutingStrategy`
`target_agents`	List of agent names that can handle matching requests TYPE: `List[str]`
`priority`	Rule priority (higher = evaluated first) TYPE: `int`
`load_balance`	Strategy for selecting among multiple targets TYPE: `LoadBalanceStrategy`
`weights`	Optional weights for weighted load balancing TYPE: `Optional[Dict[str, float]]`
`transform_request`	Optional async function to transform request before sending TYPE: `Optional[TransformFunc]`
`transform_response`	Optional async function to transform response before returning TYPE: `Optional[ResponseTransformFunc]`
`enabled`	Whether this rule is active TYPE: `bool`
`metadata`	Additional metadata for the rule TYPE: `Dict[str, Any]`

RoutingResult `dataclass` ¶

RoutingResult(matched: bool, rule: Optional[RoutingRule] = None, target_agent: Optional[RegisteredAgent] = None, match_details: Dict[str, Any] = dict())

Result of a routing decision.

ProxyStats `dataclass` ¶

ProxyStats(requests_total: int = 0, requests_routed: int = 0, requests_failed: int = 0, requests_no_match: int = 0, total_latency_ms: float = 0.0, agents_used: Dict[str, int] = dict(), rules_matched: Dict[str, int] = dict(), last_request_time: Optional[datetime] = None)

Statistics for the proxy router.

avg_latency_ms `property` ¶

avg_latency_ms: float

Average latency in milliseconds.

to_dict ¶

to_dict() -> Dict[str, Any]

Convert stats to dictionary.

Source code in packages/ai-parrot/src/parrot/a2a/router.py

def to_dict(self) -> Dict[str, Any]:
    """Convert stats to dictionary."""
    return {
        "requests_total": self.requests_total,
        "requests_routed": self.requests_routed,
        "requests_failed": self.requests_failed,
        "requests_no_match": self.requests_no_match,
        "avg_latency_ms": round(self.avg_latency_ms, 2),
        "agents_used": dict(self.agents_used),
        "rules_matched": dict(self.rules_matched),
        "last_request_time": self.last_request_time.isoformat() if self.last_request_time else None,  # noqa
    }

A2AOrchestrator ¶

A2AOrchestrator(mesh: A2AMeshDiscovery, *, name: str = 'A2AOrchestrator', default_mode: OrchestrationMode = OrchestrationMode.HYBRID, default_timeout: float = 60.0, max_parallel_agents: int = 10, max_sequential_agents: int = 5, aggregate_parallel_responses: bool = True)

Hybrid orchestrator combining rule-based routing with LLM decision-making.

This orchestrator provides intelligent request routing and multi-agent coordination. It uses deterministic rules when patterns are known and falls back to LLM-based reasoning for complex or ambiguous requests.

Architecture

Rules Engine (via A2AProxyRouter): Fast, deterministic routing
LLM Decision Engine: Complex reasoning about agent selection
Execution Engine: Parallel/sequential agent invocation
Aggregation Engine: Combine responses from multiple agents

Usage Patterns

RULES_ONLY: Pure rule-based routing, no LLM cost
LLM_ONLY: Always use LLM for decisions
HYBRID: Rules first, LLM fallback (recommended)
PARALLEL: Fan-out to multiple agents
SEQUENTIAL: Pipeline execution

Example

orchestrator = A2AOrchestrator(mesh, default_mode=OrchestrationMode.HYBRID)

Add rules (fast path)¶

orchestrator.route_by_skill("analysis", "AnalystBot") orchestrator.route_by_regex(r"urgent", "PriorityBot")

Set LLM fallback¶

orchestrator.set_fallback_llm(claude_client)

Simple case: uses rules¶

result = await orchestrator.run("Analyze this data")

Complex case: LLM decides¶

result = await orchestrator.run( "Compare our Q3 performance with competitors and suggest improvements" )

Initialize the orchestrator.

PARAMETER	DESCRIPTION
`mesh`	A2AMeshDiscovery instance for agent lookup TYPE: `A2AMeshDiscovery`
`name`	Name for this orchestrator TYPE: `str` DEFAULT: `'A2AOrchestrator'`
`default_mode`	Default orchestration mode TYPE: `OrchestrationMode` DEFAULT: `HYBRID`
`default_timeout`	Default timeout for agent calls TYPE: `float` DEFAULT: `60.0`
`max_parallel_agents`	Maximum agents in parallel execution TYPE: `int` DEFAULT: `10`
`max_sequential_agents`	Maximum agents in sequential pipeline TYPE: `int` DEFAULT: `5`
`aggregate_parallel_responses`	Whether to combine parallel responses TYPE: `bool` DEFAULT: `True`

Source code in packages/ai-parrot/src/parrot/a2a/orchestrator.py

def __init__(
    self,
    mesh: A2AMeshDiscovery,
    *,
    name: str = "A2AOrchestrator",
    default_mode: OrchestrationMode = OrchestrationMode.HYBRID,
    default_timeout: float = 60.0,
    max_parallel_agents: int = 10,
    max_sequential_agents: int = 5,
    aggregate_parallel_responses: bool = True,
):
    """
    Initialize the orchestrator.

    Args:
        mesh: A2AMeshDiscovery instance for agent lookup
        name: Name for this orchestrator
        default_mode: Default orchestration mode
        default_timeout: Default timeout for agent calls
        max_parallel_agents: Maximum agents in parallel execution
        max_sequential_agents: Maximum agents in sequential pipeline
        aggregate_parallel_responses: Whether to combine parallel responses
    """
    self.mesh = mesh
    self.name = name
    self.default_mode = default_mode
    self.default_timeout = default_timeout
    self.max_parallel_agents = max_parallel_agents
    self.max_sequential_agents = max_sequential_agents
    self.aggregate_parallel_responses = aggregate_parallel_responses

    # Internal router for rule-based routing
    self._router = A2AProxyRouter(
        mesh,
        name=f"{name}_router",
        aggregate_skills=False,  # We handle aggregation
    )

    # LLM for complex decisions
    self._llm_client: Optional["AbstractClient"] = None
    self._decision_prompt: Optional[str] = None
    self._decision_model: Optional[str] = None

    # Client cache
    self._client_cache: Dict[str, A2AClient] = {}

    # Statistics
    self._stats = OrchestratorStats()

    self.logger = logging.getLogger(f"A2A.Orchestrator.{name}")

stats `property` ¶

stats: OrchestratorStats

Get current statistics.

router `property` ¶

router: A2AProxyRouter

Access the internal router for advanced configuration.

route_by_skill ¶

route_by_skill(skill_id: str, target: Union[str, List[str]], **kwargs) -> 'A2AOrchestrator'

Add skill-based routing rule.

Source code in packages/ai-parrot/src/parrot/a2a/orchestrator.py

def route_by_skill(
    self,
    skill_id: str,
    target: Union[str, List[str]],
    **kwargs
) -> "A2AOrchestrator":
    """Add skill-based routing rule."""
    self._router.route_by_skill(skill_id, target, **kwargs)
    return self

route_by_tag ¶

route_by_tag(tag: str, target: Union[str, List[str]], **kwargs) -> 'A2AOrchestrator'

Add tag-based routing rule.

Source code in packages/ai-parrot/src/parrot/a2a/orchestrator.py

def route_by_tag(
    self,
    tag: str,
    target: Union[str, List[str]],
    **kwargs
) -> "A2AOrchestrator":
    """Add tag-based routing rule."""
    self._router.route_by_tag(tag, target, **kwargs)
    return self

route_by_regex ¶

route_by_regex(pattern: str, target: Union[str, List[str]], **kwargs) -> 'A2AOrchestrator'

Add regex-based routing rule.

Source code in packages/ai-parrot/src/parrot/a2a/orchestrator.py

def route_by_regex(
    self,
    pattern: str,
    target: Union[str, List[str]],
    **kwargs
) -> "A2AOrchestrator":
    """Add regex-based routing rule."""
    self._router.route_by_regex(pattern, target, **kwargs)
    return self

set_default ¶

set_default(agent_name: str) -> 'A2AOrchestrator'

Set default agent for unmatched requests.

Source code in packages/ai-parrot/src/parrot/a2a/orchestrator.py

def set_default(self, agent_name: str) -> "A2AOrchestrator":
    """Set default agent for unmatched requests."""
    self._router.set_default(agent_name)
    return self

set_fallback_llm ¶

set_fallback_llm(llm_client: 'AbstractClient', *, decision_prompt: Optional[str] = None, model: Optional[str] = None) -> 'A2AOrchestrator'

Configure LLM for complex orchestration decisions.

The LLM is used when: - No routing rule matches (in HYBRID mode) - Explicitly requested (LLM_ONLY mode) - Task requires multiple agents

PARAMETER	DESCRIPTION
`llm_client`	Parrot AbstractClient instance (Claude, GPT, etc.) TYPE: `'AbstractClient'`
`decision_prompt`	Custom prompt template for decisions TYPE: `Optional[str]` DEFAULT: `None`
`model`	Specific model to use for decisions TYPE: `Optional[str]` DEFAULT: `None`

RETURNS	DESCRIPTION
`'A2AOrchestrator'`	Self for method chaining

Source code in packages/ai-parrot/src/parrot/a2a/orchestrator.py

def set_fallback_llm(
    self,
    llm_client: "AbstractClient",
    *,
    decision_prompt: Optional[str] = None,
    model: Optional[str] = None,
) -> "A2AOrchestrator":
    """
    Configure LLM for complex orchestration decisions.

    The LLM is used when:
    - No routing rule matches (in HYBRID mode)
    - Explicitly requested (LLM_ONLY mode)
    - Task requires multiple agents

    Args:
        llm_client: Parrot AbstractClient instance (Claude, GPT, etc.)
        decision_prompt: Custom prompt template for decisions
        model: Specific model to use for decisions

    Returns:
        Self for method chaining
    """
    self._llm_client = llm_client
    self._decision_prompt = decision_prompt
    self._decision_model = model
    self.logger.info("LLM fallback configured")
    return self

clear_llm ¶

clear_llm() -> 'A2AOrchestrator'

Remove LLM fallback configuration.

Source code in packages/ai-parrot/src/parrot/a2a/orchestrator.py

def clear_llm(self) -> "A2AOrchestrator":
    """Remove LLM fallback configuration."""
    self._llm_client = None
    self._decision_prompt = None
    return self

run `async` ¶

run(message: str, *, mode: Optional[OrchestrationMode] = None, agents: Optional[List[str]] = None, skill_id: Optional[str] = None, tags: Optional[List[str]] = None, context_id: Optional[str] = None, timeout: Optional[float] = None, metadata: Optional[Dict[str, Any]] = None) -> OrchestrationResult

Execute orchestration for a message.

This is the main entry point. The orchestrator will: 1. Try rule-based routing if applicable 2. Fall back to LLM decision if needed (HYBRID mode) 3. Execute on selected agent(s) 4. Aggregate results if multiple agents used

PARAMETER	DESCRIPTION
`message`	The message/request to process TYPE: `str`
`mode`	Override default orchestration mode TYPE: `Optional[OrchestrationMode]` DEFAULT: `None`
`agents`	Explicit list of agents to use (bypasses routing) TYPE: `Optional[List[str]]` DEFAULT: `None`
`skill_id`	Optional skill ID hint for routing TYPE: `Optional[str]` DEFAULT: `None`
`tags`	Optional tags for routing TYPE: `Optional[List[str]]` DEFAULT: `None`
`context_id`	Optional context for multi-turn conversations TYPE: `Optional[str]` DEFAULT: `None`
`timeout`	Override default timeout TYPE: `Optional[float]` DEFAULT: `None`
`metadata`	Additional metadata TYPE: `Optional[Dict[str, Any]]` DEFAULT: `None`

RETURNS	DESCRIPTION
`OrchestrationResult`	OrchestrationResult with response(s) and execution details

Source code in packages/ai-parrot/src/parrot/a2a/orchestrator.py

async def run(
    self,
    message: str,
    *,
    mode: Optional[OrchestrationMode] = None,
    agents: Optional[List[str]] = None,
    skill_id: Optional[str] = None,
    tags: Optional[List[str]] = None,
    context_id: Optional[str] = None,
    timeout: Optional[float] = None,
    metadata: Optional[Dict[str, Any]] = None,
) -> OrchestrationResult:
    """
    Execute orchestration for a message.

    This is the main entry point. The orchestrator will:
    1. Try rule-based routing if applicable
    2. Fall back to LLM decision if needed (HYBRID mode)
    3. Execute on selected agent(s)
    4. Aggregate results if multiple agents used

    Args:
        message: The message/request to process
        mode: Override default orchestration mode
        agents: Explicit list of agents to use (bypasses routing)
        skill_id: Optional skill ID hint for routing
        tags: Optional tags for routing
        context_id: Optional context for multi-turn conversations
        timeout: Override default timeout
        metadata: Additional metadata

    Returns:
        OrchestrationResult with response(s) and execution details
    """
    start_time = time.monotonic()
    mode = mode or self.default_mode
    timeout = timeout or self.default_timeout

    self._stats.total_requests += 1
    self._stats.mode_usage[mode.value] = self._stats.mode_usage.get(mode.value, 0) + 1

    self.logger.debug(f"Orchestrating: mode={mode.value}, message={message[:100]}...")

    try:
        # If explicit agents provided, use them directly
        if agents:
            if len(agents) == 1:
                result = await self._execute_single(
                    message, agents[0], context_id, timeout
                )
            elif mode == OrchestrationMode.SEQUENTIAL:
                result = await self._execute_sequential(
                    message, agents, context_id, timeout
                )
            else:
                result = await self._execute_parallel(
                    message, agents, context_id, timeout
                )

        # Route based on mode
        elif mode == OrchestrationMode.RULES_ONLY:
            result = await self._execute_rules_only(
                message, skill_id, tags, context_id, timeout
            )

        elif mode == OrchestrationMode.LLM_ONLY:
            result = await self._execute_llm_decision(
                message, context_id, timeout
            )

        elif mode == OrchestrationMode.HYBRID:
            result = await self._execute_hybrid(
                message, skill_id, tags, context_id, timeout
            )

        elif mode == OrchestrationMode.PARALLEL:
            # Parallel to all healthy agents
            all_agents = [a.card.name for a in self.mesh.list_healthy()]
            result = await self._execute_parallel(
                message, all_agents[:self.max_parallel_agents], context_id, timeout
            )

        elif mode == OrchestrationMode.SEQUENTIAL:
            # LLM decides the sequence
            result = await self._execute_llm_decision(
                message, context_id, timeout, force_sequential=True
            )

        elif mode == OrchestrationMode.CONSENSUS:
            result = await self._execute_consensus(
                message, context_id, timeout
            )

        elif mode == OrchestrationMode.FIRST_SUCCESS:
            result = await self._execute_first_success(
                message, context_id, timeout
            )

        else:
            raise ValueError(f"Unknown orchestration mode: {mode}")

        # Update stats
        result.total_time_ms = (time.monotonic() - start_time) * 1000
        result.mode_used = mode

        if result.success:
            self._stats.successful_requests += 1
            self._stats.total_latency_ms += result.total_time_ms
        else:
            self._stats.failed_requests += 1

        for agent_name in result.agents_used:
            self._stats.agents_called[agent_name] = \
                self._stats.agents_called.get(agent_name, 0) + 1

        return result

    except Exception as e:
        self.logger.error(f"Orchestration failed: {e}", exc_info=True)
        self._stats.failed_requests += 1

        return OrchestrationResult(
            success=False,
            mode_used=mode,
            agents_used=[],
            responses=[],
            total_time_ms=(time.monotonic() - start_time) * 1000,
            metadata={"error": str(e)},
        )

close_clients `async` ¶

close_clients() -> None

Close all cached clients.

Source code in packages/ai-parrot/src/parrot/a2a/orchestrator.py

async def close_clients(self) -> None:
    """Close all cached clients."""
    for client in self._client_cache.values():
        try:
            await client.disconnect()
        except Exception:
            pass
    self._client_cache.clear()

ask `async` ¶

ask(message: str, *, agent: Optional[str] = None, mode: Optional[OrchestrationMode] = None, **kwargs) -> str

Shortcut: get response as string.

PARAMETER	DESCRIPTION
`message`	Message to send TYPE: `str`
`agent`	Optional specific agent TYPE: `Optional[str]` DEFAULT: `None`
`mode`	Optional mode override TYPE: `Optional[OrchestrationMode]` DEFAULT: `None`
`**kwargs`	Additional arguments DEFAULT: `{}`

RETURNS	DESCRIPTION
`str`	Response text

RAISES	DESCRIPTION
`RuntimeError`	If orchestration fails

Source code in packages/ai-parrot/src/parrot/a2a/orchestrator.py

async def ask(
    self,
    message: str,
    *,
    agent: Optional[str] = None,
    mode: Optional[OrchestrationMode] = None,
    **kwargs
) -> str:
    """
    Shortcut: get response as string.

    Args:
        message: Message to send
        agent: Optional specific agent
        mode: Optional mode override
        **kwargs: Additional arguments

    Returns:
        Response text

    Raises:
        RuntimeError: If orchestration fails
    """
    if agent:
        result = await self.run(message, agents=[agent], **kwargs)
    else:
        result = await self.run(message, mode=mode, **kwargs)

    if not result.success:
        error = result.metadata.get("error", "Unknown error")
        raise RuntimeError(f"Orchestration failed: {error}")

    return result.primary_response or ""

fan_out `async` ¶

fan_out(message: str, agents: List[str], **kwargs) -> Dict[str, Union[str, Exception]]

Send to multiple agents and collect responses.

PARAMETER	DESCRIPTION
`message`	Message to send TYPE: `str`
`agents`	List of agent names TYPE: `List[str]`
`**kwargs`	Additional arguments DEFAULT: `{}`

RETURNS	DESCRIPTION
`Dict[str, Union[str, Exception]]`	Dict mapping agent name to response or exception

Source code in packages/ai-parrot/src/parrot/a2a/orchestrator.py

async def fan_out(
    self,
    message: str,
    agents: List[str],
    **kwargs
) -> Dict[str, Union[str, Exception]]:
    """
    Send to multiple agents and collect responses.

    Args:
        message: Message to send
        agents: List of agent names
        **kwargs: Additional arguments

    Returns:
        Dict mapping agent name to response or exception
    """
    result = await self.run(
        message,
        agents=agents,
        mode=OrchestrationMode.PARALLEL,
        **kwargs
    )

    return {
        r.agent_name: r.response if r.success else Exception(r.error or "Failed")
        for r in result.responses
    }

pipeline `async` ¶

pipeline(message: str, agents: List[str], **kwargs) -> str

Execute sequential pipeline.

PARAMETER	DESCRIPTION
`message`	Initial message TYPE: `str`
`agents`	Ordered list of agents TYPE: `List[str]`
`**kwargs`	Additional arguments DEFAULT: `{}`

RETURNS	DESCRIPTION
`str`	Final output from last agent

Source code in packages/ai-parrot/src/parrot/a2a/orchestrator.py

async def pipeline(
    self,
    message: str,
    agents: List[str],
    **kwargs
) -> str:
    """
    Execute sequential pipeline.

    Args:
        message: Initial message
        agents: Ordered list of agents
        **kwargs: Additional arguments

    Returns:
        Final output from last agent
    """
    result = await self.run(
        message,
        agents=agents,
        mode=OrchestrationMode.SEQUENTIAL,
        **kwargs
    )

    if not result.success:
        error = result.metadata.get("error", "Pipeline failed")
        raise RuntimeError(error)

    return result.final_output or ""

get_info ¶

get_info() -> Dict[str, Any]

Get orchestrator state information.

Source code in packages/ai-parrot/src/parrot/a2a/orchestrator.py

def get_info(self) -> Dict[str, Any]:
    """Get orchestrator state information."""
    return {
        "name": self.name,
        "default_mode": self.default_mode.value,
        "llm_configured": self._llm_client is not None,
        "rules_count": len(self._router._rules),
        "default_agent": self._router._default_agent,
        "mesh_agents": len(self.mesh),
        "mesh_healthy": len(self.mesh.list_healthy()),
        "cached_clients": len(self._client_cache),
        "stats": self._stats.to_dict(),
    }

OrchestrationMode ¶

Bases: str, Enum

Mode of orchestration.

LLMDecisionStrategy ¶

Bases: str, Enum

Strategy for LLM-based agent selection.

OrchestrationPlan ¶

Bases: BaseModel

Complete orchestration plan from LLM.

OrchestrationResult `dataclass` ¶

OrchestrationResult(success: bool, mode_used: OrchestrationMode, agents_used: List[str], responses: List[AgentExecutionResult], final_output: Optional[str] = None, total_time_ms: float = 0.0, llm_fallback_used: bool = False, llm_decision: Optional[OrchestrationPlan] = None, metadata: Dict[str, Any] = dict())

Complete result from orchestration.

primary_response `property` ¶

primary_response: Optional[str]

Get the primary response (first successful or final output).

to_dict ¶

to_dict() -> Dict[str, Any]

Convert to dictionary.

Source code in packages/ai-parrot/src/parrot/a2a/orchestrator.py

def to_dict(self) -> Dict[str, Any]:
    """Convert to dictionary."""
    return {
        "success": self.success,
        "mode_used": self.mode_used.value,
        "agents_used": self.agents_used,
        "responses": [
            {
                "agent": r.agent_name,
                "success": r.success,
                "response": r.response[:500] if r.response else None,
                "error": r.error,
                "latency_ms": r.latency_ms,
            }
            for r in self.responses
        ],
        "final_output": self.final_output[:1000] if self.final_output else None,
        "total_time_ms": round(self.total_time_ms, 2),
        "llm_fallback_used": self.llm_fallback_used,
        "llm_decision": self.llm_decision.model_dump() if self.llm_decision else None,
    }

AgentExecutionResult `dataclass` ¶

AgentExecutionResult(agent_name: str, success: bool, response: Optional[str] = None, task: Optional[Task] = None, error: Optional[str] = None, latency_ms: float = 0.0, metadata: Dict[str, Any] = dict())

Result from a single agent execution.

OrchestratorStats `dataclass` ¶

OrchestratorStats(total_requests: int = 0, successful_requests: int = 0, failed_requests: int = 0, rules_used: int = 0, llm_fallback_used: int = 0, parallel_executions: int = 0, sequential_executions: int = 0, total_latency_ms: float = 0.0, agents_called: Dict[str, int] = dict(), mode_usage: Dict[str, int] = dict())

Statistics for the orchestrator.

parse_task_state ¶

parse_task_state(value: Union[str, TaskState]) -> TaskState

Parse a TaskState from either the v0.3 or the v1.0 format.

Source code in packages/ai-parrot/src/parrot/a2a/models.py

def parse_task_state(value: Union[str, "TaskState"]) -> "TaskState":
    """Parse a TaskState from either the v0.3 or the v1.0 format."""
    if isinstance(value, TaskState):
        return value
    if value in _TASK_STATE_COMPAT:
        return _TASK_STATE_COMPAT[value]
    return TaskState(value)

parse_role ¶

parse_role(value: Union[str, Role]) -> Role

Parse a Role from either the v0.3 or the v1.0 format.

Source code in packages/ai-parrot/src/parrot/a2a/models.py

def parse_role(value: Union[str, "Role"]) -> "Role":
    """Parse a Role from either the v0.3 or the v1.0 format."""
    if isinstance(value, Role):
        return value
    if value in _ROLE_COMPAT:
        return _ROLE_COMPAT[value]
    return Role(value)

serialize_task_state ¶

serialize_task_state(state: TaskState, version: str = '1.0') -> str

Serialize a TaskState to the wire value for the target protocol version.

Source code in packages/ai-parrot/src/parrot/a2a/models.py

def serialize_task_state(state: "TaskState", version: str = "1.0") -> str:
    """Serialize a TaskState to the wire value for the target protocol version."""
    if version == "0.3":
        return _TASK_STATE_V03.get(state, state.value)
    return state.value

serialize_role ¶

serialize_role(role: Role, version: str = '1.0') -> str

Serialize a Role to the wire value for the target protocol version.

Source code in packages/ai-parrot/src/parrot/a2a/models.py

def serialize_role(role: "Role", version: str = "1.0") -> str:
    """Serialize a Role to the wire value for the target protocol version."""
    if version == "0.3":
        return _ROLE_V03.get(role, role.value)
    return role.value

A2A (Agent-to-Agent)¶

A2AClient ¶

agent_card property ¶

connect async ¶

disconnect async ¶

discover async ¶

get_skills ¶

get_skill ¶

send_message async ¶

stream_message async ¶

invoke_skill async ¶

get_task async ¶

list_tasks async ¶

cancel_task async ¶

create_push_config async ¶

get_push_config async ¶

list_push_configs async ¶

delete_push_config async ¶

rpc_call async ¶

A2AAgentConnection dataclass ¶

A2ARemoteAgentTool ¶

clone ¶

A2ARemoteSkillTool ¶

clone ¶

A2AClientMixin ¶

Option 1: Connect to remote agents directly¶

Now the agent can use remote agents as tools¶

Or call remote agents directly¶

Option 2: Use mesh discovery¶

Option 3: Use router for rule-based routing¶

Option 4: Use orchestrator for hybrid routing¶

set_matrix_transport ¶

get_matrix_transport ¶

set_mesh ¶

get_mesh ¶

discover_from_mesh async ¶

Connect to all agents with data analysis skill¶

Connect to all agents tagged as "support"¶

Connect to all healthy agents¶

set_router ¶

get_router ¶

route_to_agent async ¶

set_orchestrator ¶

get_orchestrator ¶

orchestrate async ¶

fan_out async ¶

pipeline async ¶

add_a2a_agent async ¶

remove_a2a_agent async ¶

list_a2a_agents ¶

get_a2a_agent ¶

get_a2a_client ¶

ask_remote_agent async ¶

invoke_remote_skill async ¶

shutdown_a2a async ¶

shutdown async ¶

AgentCard dataclass ¶

url property writable ¶

preferred_transport property writable ¶

protocol_version property ¶

AgentSkill dataclass ¶

AgentCapabilities dataclass ¶

Task dataclass ¶

TaskState ¶

TaskStatus dataclass ¶

Message dataclass ¶

get_text ¶

get_data ¶

Part dataclass ¶

Artifact dataclass ¶

from_response classmethod ¶

RegisteredAgent dataclass ¶

AgentConfig dataclass ¶

AgentInterface dataclass ¶

AgentProvider dataclass ¶

AgentExtension dataclass ¶

AgentCardSignature dataclass ¶

SecurityScheme dataclass ¶

APIKeySecurityScheme dataclass ¶

HTTPAuthSecurityScheme dataclass ¶

agent_card `property` ¶

connect `async` ¶

disconnect `async` ¶

discover `async` ¶

send_message `async` ¶

stream_message `async` ¶

invoke_skill `async` ¶

get_task `async` ¶

list_tasks `async` ¶

cancel_task `async` ¶

create_push_config `async` ¶

get_push_config `async` ¶

list_push_configs `async` ¶

delete_push_config `async` ¶

rpc_call `async` ¶

A2AAgentConnection `dataclass` ¶

discover_from_mesh `async` ¶

route_to_agent `async` ¶

orchestrate `async` ¶

fan_out `async` ¶

pipeline `async` ¶

add_a2a_agent `async` ¶

remove_a2a_agent `async` ¶

ask_remote_agent `async` ¶

invoke_remote_skill `async` ¶

shutdown_a2a `async` ¶

shutdown `async` ¶

AgentCard `dataclass` ¶

url `property` `writable` ¶

preferred_transport `property` `writable` ¶

protocol_version `property` ¶

AgentSkill `dataclass` ¶

AgentCapabilities `dataclass` ¶

Task `dataclass` ¶

TaskStatus `dataclass` ¶

Message `dataclass` ¶

Part `dataclass` ¶

Artifact `dataclass` ¶

from_response `classmethod` ¶

RegisteredAgent `dataclass` ¶

AgentConfig `dataclass` ¶

AgentInterface `dataclass` ¶

AgentProvider `dataclass` ¶

AgentExtension `dataclass` ¶

AgentCardSignature `dataclass` ¶

SecurityScheme `dataclass` ¶

APIKeySecurityScheme `dataclass` ¶

HTTPAuthSecurityScheme `dataclass` ¶

OAuth2SecurityScheme `dataclass` ¶

OpenIdConnectSecurityScheme `dataclass` ¶

MutualTlsSecurityScheme `dataclass` ¶

SecurityRequirement `dataclass` ¶

SendMessageConfiguration `dataclass` ¶

TaskPushNotificationConfig `dataclass` ¶

AuthenticationInfo `dataclass` ¶

A2AError `dataclass` ¶

stats `property` ¶

from_config `classmethod` ¶

start `async` ¶

stop `async` ¶

register `async` ¶

unregister `async` ¶

check_health_now `async` ¶

A2AEndpoint `dataclass` ¶

DiscoveryStats `dataclass` ¶

stats `property` ¶

route_message `async` ¶

route_message_stream `async` ¶

invoke_skill `async` ¶

close_clients `async` ¶

ask `async` ¶

fan_out `async` ¶

pipeline `async` ¶