Skip to content

Agent-to-Agent

Protocol Types

A2AMessage dataclass

A message in the A2A protocol.

Attributes:

Name Type Description
message_type A2AMessageType

The type of this message.
sender str

Identifier of the sending agent.
receiver str

Identifier of the receiving agent.
payload dict[str, Any]

The message payload data.
correlation_id str

ID linking related messages in a conversation.
metadata dict[str, Any]

Additional metadata.
Source code in src/agenticapi/interface/a2a/protocol.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
@dataclass(frozen=True, slots=True)
class A2AMessage:
    """A message in the A2A protocol.

    Attributes:
        message_type: The type of this message.
        sender: Identifier of the sending agent.
        receiver: Identifier of the receiving agent.
        payload: The message payload data.
        correlation_id: ID linking related messages in a conversation.
        metadata: Additional metadata.
    """

    message_type: A2AMessageType
    sender: str
    receiver: str
    payload: dict[str, Any] = field(default_factory=dict)
    correlation_id: str = ""
    metadata: dict[str, Any] = field(default_factory=dict)

A2AMessageType

Bases: StrEnum

Standard A2A protocol message types.

Source code in src/agenticapi/interface/a2a/protocol.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
class A2AMessageType(StrEnum):
    """Standard A2A protocol message types."""

    DISCOVER = "discover"
    INTENT = "intent"
    NEGOTIATE = "negotiate"
    DELEGATE = "delegate"
    OBSERVE = "observe"
    REVISE = "revise"
    EXPLAIN = "explain"
    VERIFY = "verify"
    RESPONSE = "response"
    ERROR = "error"

A2ARequest dataclass

A request sent to an A2A service.

Attributes:

Name Type Description
capability_name str

The capability being invoked.
parameters dict[str, Any]

Request parameters.
sender str

Identifier of the requesting agent.
correlation_id str

ID for tracking the conversation.
timeout_seconds float

Maximum time to wait for a response.
Source code in src/agenticapi/interface/a2a/protocol.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
@dataclass(frozen=True, slots=True)
class A2ARequest:
    """A request sent to an A2A service.

    Attributes:
        capability_name: The capability being invoked.
        parameters: Request parameters.
        sender: Identifier of the requesting agent.
        correlation_id: ID for tracking the conversation.
        timeout_seconds: Maximum time to wait for a response.
    """

    capability_name: str
    parameters: dict[str, Any] = field(default_factory=dict)
    sender: str = ""
    correlation_id: str = ""
    timeout_seconds: float = 30.0

A2AResponse dataclass

A response from an A2A service.

Attributes:

Name Type Description
success bool

Whether the request was processed successfully.
result Any

The result data.
error str | None

Error message if unsuccessful.
correlation_id str

ID linking to the original request.
Source code in src/agenticapi/interface/a2a/protocol.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
@dataclass(frozen=True, slots=True)
class A2AResponse:
    """A response from an A2A service.

    Attributes:
        success: Whether the request was processed successfully.
        result: The result data.
        error: Error message if unsuccessful.
        correlation_id: ID linking to the original request.
    """

    success: bool
    result: Any = None
    error: str | None = None
    correlation_id: str = ""

Capability

Capability dataclass

A capability offered by an A2A service.

Attributes:

Name Type Description
name str

Unique name identifying this capability.
description str

Human-readable description of what this capability does.
input_schema dict[str, Any]

JSON-schema-like dict describing expected input.
output_schema dict[str, Any]

JSON-schema-like dict describing output format.
sla_max_latency_ms int

Maximum expected latency in milliseconds.
sla_availability float

Target availability (0.0-1.0).
Source code in src/agenticapi/interface/a2a/capability.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
@dataclass(frozen=True, slots=True)
class Capability:
    """A capability offered by an A2A service.

    Attributes:
        name: Unique name identifying this capability.
        description: Human-readable description of what this capability does.
        input_schema: JSON-schema-like dict describing expected input.
        output_schema: JSON-schema-like dict describing output format.
        sla_max_latency_ms: Maximum expected latency in milliseconds.
        sla_availability: Target availability (0.0-1.0).
    """

    name: str
    description: str = ""
    input_schema: dict[str, Any] = field(default_factory=dict)
    output_schema: dict[str, Any] = field(default_factory=dict)
    sla_max_latency_ms: int = 5000
    sla_availability: float = 0.99

CapabilityRegistry

Registry of capabilities offered by a service.

Agents register their capabilities here, and remote agents can discover them via the A2A DISCOVER message.

Example

registry = CapabilityRegistry() registry.register(Capability( name="inventory_lookup", description="Look up current inventory levels", )) caps = registry.list_capabilities()

Source code in src/agenticapi/interface/a2a/capability.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
class CapabilityRegistry:
    """Registry of capabilities offered by a service.

    Agents register their capabilities here, and remote agents
    can discover them via the A2A DISCOVER message.

    Example:
        registry = CapabilityRegistry()
        registry.register(Capability(
            name="inventory_lookup",
            description="Look up current inventory levels",
        ))
        caps = registry.list_capabilities()
    """

    def __init__(self) -> None:
        self._capabilities: dict[str, Capability] = {}

    def register(self, capability: Capability) -> None:
        """Register a capability.

        Args:
            capability: The capability to register.
        """
        self._capabilities[capability.name] = capability
        logger.info("a2a_capability_registered", capability=capability.name)

    def get(self, name: str) -> Capability | None:
        """Look up a capability by name.

        Args:
            name: The capability name.

        Returns:
            The capability if found, None otherwise.
        """
        return self._capabilities.get(name)

    def list_capabilities(self) -> list[Capability]:
        """List all registered capabilities.

        Returns:
            All registered capabilities.
        """
        return list(self._capabilities.values())

    def has(self, name: str) -> bool:
        """Check if a capability is registered.

        Args:
            name: The capability name.

        Returns:
            True if the capability exists.
        """
        return name in self._capabilities

register 

register(capability: Capability) -> None

Register a capability.

Parameters:

Name Type Description Default
capability Capability

The capability to register.

 required
Source code in src/agenticapi/interface/a2a/capability.py 
56
57
58
59
60
61
62
63
def register(self, capability: Capability) -> None:
    """Register a capability.

    Args:
        capability: The capability to register.
    """
    self._capabilities[capability.name] = capability
    logger.info("a2a_capability_registered", capability=capability.name)

get 

get(name: str) -> Capability | None

Look up a capability by name.

Parameters:

Name Type Description Default
name str

The capability name.

 required

Returns:

Type Description
Capability | None

The capability if found, None otherwise.
Source code in src/agenticapi/interface/a2a/capability.py 
65
66
67
68
69
70
71
72
73
74
def get(self, name: str) -> Capability | None:
    """Look up a capability by name.

    Args:
        name: The capability name.

    Returns:
        The capability if found, None otherwise.
    """
    return self._capabilities.get(name)

list_capabilities 

list_capabilities() -> list[Capability]

List all registered capabilities.

Returns:

Type Description
list[Capability]

All registered capabilities.
Source code in src/agenticapi/interface/a2a/capability.py 
76
77
78
79
80
81
82
def list_capabilities(self) -> list[Capability]:
    """List all registered capabilities.

    Returns:
        All registered capabilities.
    """
    return list(self._capabilities.values())

has 

has(name: str) -> bool

Check if a capability is registered.

Parameters:

Name Type Description Default
name str

The capability name.

 required

Returns:

Type Description
bool

True if the capability exists.
Source code in src/agenticapi/interface/a2a/capability.py 
84
85
86
87
88
89
90
91
92
93
def has(self, name: str) -> bool:
    """Check if a capability is registered.

    Args:
        name: The capability name.

    Returns:
        True if the capability exists.
    """
    return name in self._capabilities

Trust

TrustPolicy dataclass

Policy governing trust between agents.

Attributes:

Name Type Description
initial_trust float

Trust score for unknown agents (0.0-1.0).
min_trust_for_read float

Minimum trust to allow read operations.
min_trust_for_write float

Minimum trust to allow write operations.
decay_per_failure float

How much trust decreases on failure.
gain_per_success float

How much trust increases on success.
Source code in src/agenticapi/interface/a2a/trust.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
@dataclass(frozen=True, slots=True)
class TrustPolicy:
    """Policy governing trust between agents.

    Attributes:
        initial_trust: Trust score for unknown agents (0.0-1.0).
        min_trust_for_read: Minimum trust to allow read operations.
        min_trust_for_write: Minimum trust to allow write operations.
        decay_per_failure: How much trust decreases on failure.
        gain_per_success: How much trust increases on success.
    """

    initial_trust: float = 0.5
    min_trust_for_read: float = 0.3
    min_trust_for_write: float = 0.8
    decay_per_failure: float = 0.1
    gain_per_success: float = 0.05

TrustScorer

Tracks and computes trust scores for remote agents.

Trust scores are updated based on interaction outcomes and used to gate what operations remote agents can perform.

Example

scorer = TrustScorer(policy=TrustPolicy()) score = scorer.get_score("agent-123") scorer.record_success("agent-123")

Source code in src/agenticapi/interface/a2a/trust.py 
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
class TrustScorer:
    """Tracks and computes trust scores for remote agents.

    Trust scores are updated based on interaction outcomes and
    used to gate what operations remote agents can perform.

    Example:
        scorer = TrustScorer(policy=TrustPolicy())
        score = scorer.get_score("agent-123")
        scorer.record_success("agent-123")
    """

    def __init__(self, *, policy: TrustPolicy | None = None) -> None:
        """Initialize the trust scorer.

        Args:
            policy: The trust policy to apply. Uses defaults if None.
        """
        self._policy = policy or TrustPolicy()
        self._scores: dict[str, float] = {}

    @property
    def policy(self) -> TrustPolicy:
        """The active trust policy."""
        return self._policy

    def get_score(self, agent_id: str) -> float:
        """Get the trust score for an agent.

        Args:
            agent_id: The agent identifier.

        Returns:
            The trust score (0.0-1.0). Returns initial_trust for unknown agents.
        """
        return self._scores.get(agent_id, self._policy.initial_trust)

    def can_read(self, agent_id: str) -> bool:
        """Check if an agent is trusted enough for read operations.

        Args:
            agent_id: The agent identifier.

        Returns:
            True if the agent meets the minimum read trust threshold.
        """
        return self.get_score(agent_id) >= self._policy.min_trust_for_read

    def can_write(self, agent_id: str) -> bool:
        """Check if an agent is trusted enough for write operations.

        Args:
            agent_id: The agent identifier.

        Returns:
            True if the agent meets the minimum write trust threshold.
        """
        return self.get_score(agent_id) >= self._policy.min_trust_for_write

    def record_success(self, agent_id: str) -> float:
        """Record a successful interaction, increasing trust.

        Args:
            agent_id: The agent identifier.

        Returns:
            The updated trust score.
        """
        current = self.get_score(agent_id)
        new_score = min(1.0, current + self._policy.gain_per_success)
        self._scores[agent_id] = new_score
        logger.debug("trust_updated", agent_id=agent_id, old=current, new=new_score, reason="success")
        return new_score

    def record_failure(self, agent_id: str) -> float:
        """Record a failed interaction, decreasing trust.

        Args:
            agent_id: The agent identifier.

        Returns:
            The updated trust score.
        """
        current = self.get_score(agent_id)
        new_score = max(0.0, current - self._policy.decay_per_failure)
        self._scores[agent_id] = new_score
        logger.debug("trust_updated", agent_id=agent_id, old=current, new=new_score, reason="failure")
        return new_score

policy property 

policy: TrustPolicy

The active trust policy.

__init__ 

__init__(*, policy: TrustPolicy | None = None) -> None

Initialize the trust scorer.

Parameters:

Name Type Description Default
policy TrustPolicy | None

The trust policy to apply. Uses defaults if None.

 None
Source code in src/agenticapi/interface/a2a/trust.py 
47
48
49
50
51
52
53
54
def __init__(self, *, policy: TrustPolicy | None = None) -> None:
    """Initialize the trust scorer.

    Args:
        policy: The trust policy to apply. Uses defaults if None.
    """
    self._policy = policy or TrustPolicy()
    self._scores: dict[str, float] = {}

get_score 

get_score(agent_id: str) -> float

Get the trust score for an agent.

Parameters:

Name Type Description Default
agent_id str

The agent identifier.

 required

Returns:

Type Description
float

The trust score (0.0-1.0). Returns initial_trust for unknown agents.
Source code in src/agenticapi/interface/a2a/trust.py 
61
62
63
64
65
66
67
68
69
70
def get_score(self, agent_id: str) -> float:
    """Get the trust score for an agent.

    Args:
        agent_id: The agent identifier.

    Returns:
        The trust score (0.0-1.0). Returns initial_trust for unknown agents.
    """
    return self._scores.get(agent_id, self._policy.initial_trust)

can_read 

can_read(agent_id: str) -> bool

Check if an agent is trusted enough for read operations.

Parameters:

Name Type Description Default
agent_id str

The agent identifier.

 required

Returns:

Type Description
bool

True if the agent meets the minimum read trust threshold.
Source code in src/agenticapi/interface/a2a/trust.py 
72
73
74
75
76
77
78
79
80
81
def can_read(self, agent_id: str) -> bool:
    """Check if an agent is trusted enough for read operations.

    Args:
        agent_id: The agent identifier.

    Returns:
        True if the agent meets the minimum read trust threshold.
    """
    return self.get_score(agent_id) >= self._policy.min_trust_for_read

can_write 

can_write(agent_id: str) -> bool

Check if an agent is trusted enough for write operations.

Parameters:

Name Type Description Default
agent_id str

The agent identifier.

 required

Returns:

Type Description
bool

True if the agent meets the minimum write trust threshold.
Source code in src/agenticapi/interface/a2a/trust.py 
83
84
85
86
87
88
89
90
91
92
def can_write(self, agent_id: str) -> bool:
    """Check if an agent is trusted enough for write operations.

    Args:
        agent_id: The agent identifier.

    Returns:
        True if the agent meets the minimum write trust threshold.
    """
    return self.get_score(agent_id) >= self._policy.min_trust_for_write

record_success 

record_success(agent_id: str) -> float

Record a successful interaction, increasing trust.

Parameters:

Name Type Description Default
agent_id str

The agent identifier.

 required

Returns:

Type Description
float

The updated trust score.
Source code in src/agenticapi/interface/a2a/trust.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def record_success(self, agent_id: str) -> float:
    """Record a successful interaction, increasing trust.

    Args:
        agent_id: The agent identifier.

    Returns:
        The updated trust score.
    """
    current = self.get_score(agent_id)
    new_score = min(1.0, current + self._policy.gain_per_success)
    self._scores[agent_id] = new_score
    logger.debug("trust_updated", agent_id=agent_id, old=current, new=new_score, reason="success")
    return new_score

record_failure 

record_failure(agent_id: str) -> float

Record a failed interaction, decreasing trust.

Parameters:

Name Type Description Default
agent_id str

The agent identifier.

 required

Returns:

Type Description
float

The updated trust score.
Source code in src/agenticapi/interface/a2a/trust.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def record_failure(self, agent_id: str) -> float:
    """Record a failed interaction, decreasing trust.

    Args:
        agent_id: The agent identifier.

    Returns:
        The updated trust score.
    """
    current = self.get_score(agent_id)
    new_score = max(0.0, current - self._policy.decay_per_failure)
    self._scores[agent_id] = new_score
    logger.debug("trust_updated", agent_id=agent_id, old=current, new=new_score, reason="failure")
    return new_score