A2A API

This section provides detailed API documentation for the Agent-to-Agent (A2A) protocol components in Scoras.

A2A Client

A2AClient

class A2AClient:
    def __init__(
        self,
        agent_url: str,
        api_key: Optional[str] = None,
        enable_scoring: bool = True
    ):
        """
        Initialize an A2A client to connect to A2A-compatible agents.

        Args:
            agent_url: URL of the A2A agent
            api_key: Optional API key for authentication
            enable_scoring: Whether to track complexity scoring
        """

    async def send_task(
        self,
        message: str,
        task_id: Optional[str] = None,
        metadata: Optional[Dict[str, Any]] = None
    ) -> Dict[str, Any]:
        """
        Send a task to the agent asynchronously.

        Args:
            message: Message to send
            task_id: Optional task ID (generated if not provided)
            metadata: Optional metadata for the task

        Returns:
            Task information including ID
        """

    def send_task_sync(
        self,
        message: str,
        task_id: Optional[str] = None,
        metadata: Optional[Dict[str, Any]] = None
    ) -> Dict[str, Any]:
        """
        Send a task to the agent synchronously.

        Args:
            message: Message to send
            task_id: Optional task ID (generated if not provided)
            metadata: Optional metadata for the task

        Returns:
            Task information including ID
        """

    async def wait_for_task(
        self,
        task_id: str,
        timeout: Optional[float] = None,
        poll_interval: float = 0.5
    ) -> Dict[str, Any]:
        """
        Wait for a task to complete asynchronously.

        Args:
            task_id: ID of the task to wait for
            timeout: Optional timeout in seconds
            poll_interval: Interval between polls in seconds

        Returns:
            Completed task information
        """

    def wait_for_task_sync(
        self,
        task_id: str,
        timeout: Optional[float] = None,
        poll_interval: float = 0.5
    ) -> Dict[str, Any]:
        """
        Wait for a task to complete synchronously.

        Args:
            task_id: ID of the task to wait for
            timeout: Optional timeout in seconds
            poll_interval: Interval between polls in seconds

        Returns:
            Completed task information
        """

    async def get_task_status(self, task_id: str) -> Dict[str, Any]:
        """
        Get the status of a task asynchronously.

        Args:
            task_id: ID of the task to get status for

        Returns:
            Task status information
        """

    def get_task_status_sync(self, task_id: str) -> Dict[str, Any]:
        """
        Get the status of a task synchronously.

        Args:
            task_id: ID of the task to get status for

        Returns:
            Task status information
        """

    async def list_agent_skills(self) -> List[Dict[str, Any]]:
        """
        List the skills of the agent asynchronously.

        Returns:
            List of skill descriptions
        """

    def list_agent_skills_sync(self) -> List[Dict[str, Any]]:
        """
        List the skills of the agent synchronously.

        Returns:
            List of skill descriptions
        """

    async def get_agent_info(self) -> Dict[str, Any]:
        """
        Get information about the agent asynchronously.

        Returns:
            Dictionary containing agent information
        """

    def get_agent_info_sync(self) -> Dict[str, Any]:
        """
        Get information about the agent synchronously.

        Returns:
            Dictionary containing agent information
        """

    def get_complexity_score(self) -> Dict[str, Any]:
        """
        Get the complexity score for the A2A client.

        Returns:
            Dictionary containing complexity score information
        """

A2A Server

A2AServer

class A2AServer:
    def __init__(
        self,
        name: str,
        description: str,
        agent: Agent,
        skills: List[Dict[str, Any]],
        enable_scoring: bool = True
    ):
        """
        Initialize an A2A server.

        Args:
            name: Name of the agent
            description: Description of the agent
            agent: Agent to power the server
            skills: List of skills the agent has
            enable_scoring: Whether to track complexity scoring
        """

    def add_skill(self, skill: Dict[str, Any]) -> None:
        """
        Add a skill to the agent.

        Args:
            skill: Skill description
        """

    async def handle_request(self, request: Dict[str, Any]) -> Dict[str, Any]:
        """
        Handle an A2A request asynchronously.

        Args:
            request: A2A request in JSON-RPC format

        Returns:
            A2A response in JSON-RPC format
        """

    async def handle_task(self, task: Dict[str, Any]) -> Dict[str, Any]:
        """
        Handle a task request asynchronously.

        Args:
            task: Task request

        Returns:
            Task response
        """

    def get_task(self, task_id: str) -> Optional[Dict[str, Any]]:
        """
        Get a task by ID.

        Args:
            task_id: ID of the task to get

        Returns:
            Task information if found, None otherwise
        """

    def get_complexity_score(self) -> Dict[str, Any]:
        """
        Get the complexity score for the A2A server.

        Returns:
            Dictionary containing complexity score information
        """

A2A Functions

create_agent_skill

def create_agent_skill(
    id: str,
    name: str,
    description: str,
    tags: Optional[List[str]] = None,
    complexity: str = "standard"
) -> Dict[str, Any]:
    """
    Create a skill description for an A2A agent.

    Args:
        id: Unique identifier for the skill
        name: Name of the skill
        description: Description of what the skill does
        tags: Optional list of tags for the skill
        complexity: Complexity level ("simple", "standard", "complex")

    Returns:
        Skill description as a dictionary
    """

create_a2a_server

def create_a2a_server(
    name: str,
    description: str,
    agent: Agent,
    skills: List[Dict[str, Any]],
    enable_scoring: bool = True
) -> A2AServer:
    """
    Factory function to create an A2A server.

    Args:
        name: Name of the agent
        description: Description of the agent
        agent: Agent to power the server
        skills: List of skills the agent has
        enable_scoring: Whether to track complexity scoring

    Returns:
        Instantiated A2AServer
    """

run_a2a_server

async def run_a2a_server(
    server: A2AServer,
    host: str = "0.0.0.0",
    port: int = 8000
) -> None:
    """
    Run an A2A server asynchronously.

    Args:
        server: A2A server to run
        host: Host to bind to
        port: Port to listen on
    """

run_a2a_server_sync

def run_a2a_server_sync(
    server: A2AServer,
    host: str = "0.0.0.0",
    port: int = 8000
) -> None:
    """
    Run an A2A server synchronously.

    Args:
        server: A2A server to run
        host: Host to bind to
        port: Port to listen on
    """

A2A Coordinator

A2ACoordinator

class A2ACoordinator:
    def __init__(
        self,
        name: str,
        description: str,
        agents: Dict[str, A2AClient],
        routing_strategy: str = "auto",
        enable_scoring: bool = True
    ):
        """
        Initialize an A2A coordinator that routes tasks to appropriate agents.

        Args:
            name: Name of the coordinator
            description: Description of the coordinator
            agents: Dictionary mapping agent names to A2AClient instances
            routing_strategy: Strategy for routing ("auto", "manual", "round-robin")
            enable_scoring: Whether to track complexity scoring
        """

    def add_agent(self, name: str, client: A2AClient) -> None:
        """
        Add an agent to the coordinator.

        Args:
            name: Name of the agent
            client: A2AClient instance for the agent
        """

    async def route_task(
        self,
        message: str,
        metadata: Optional[Dict[str, Any]] = None
    ) -> Dict[str, Any]:
        """
        Route a task to the appropriate agent asynchronously.

        Args:
            message: Message to route
            metadata: Optional metadata for the task

        Returns:
            Task information including ID and assigned agent
        """

    def route_task_sync(
        self,
        message: str,
        metadata: Optional[Dict[str, Any]] = None
    ) -> Dict[str, Any]:
        """
        Route a task to the appropriate agent synchronously.

        Args:
            message: Message to route
            metadata: Optional metadata for the task

        Returns:
            Task information including ID and assigned agent
        """

    async def wait_for_routed_task(
        self,
        task_id: str,
        timeout: Optional[float] = None
    ) -> Dict[str, Any]:
        """
        Wait for a routed task to complete asynchronously.

        Args:
            task_id: ID of the task to wait for
            timeout: Optional timeout in seconds

        Returns:
            Completed task information
        """

    def wait_for_routed_task_sync(
        self,
        task_id: str,
        timeout: Optional[float] = None
    ) -> Dict[str, Any]:
        """
        Wait for a routed task to complete synchronously.

        Args:
            task_id: ID of the task to wait for
            timeout: Optional timeout in seconds

        Returns:
            Completed task information
        """

    def get_complexity_score(self) -> Dict[str, Any]:
        """
        Get the complexity score for the A2A coordinator.

        Returns:
            Dictionary containing complexity score information
        """