agentconnect.communication.protocols.collaboration module

Collaboration protocol that enables dynamic capability discovery and task delegation.

This module provides the CollaborationProtocol, which implements communication patterns for peer-to-peer collaborative interactions between agents. It enables agents to discover each other’s capabilities and delegate tasks without requiring central coordination.

class RequestCapabilityPayload(capability_name=None, capability_description=None, input_schema=None, limit=10)

Bases: object

Payload for dynamically discovering capabilities from other agents.

This structure enables an agent to search for specific capabilities across the network without needing to know in advance which agents provide them.

Parameters:

  • capability_name (str | None)

  • capability_description (str | None)

  • input_schema (Dict[str, str] | None)

  • limit (int | None)

capability_name: Optional[str] = None
capability_description: Optional[str] = None
input_schema: Optional[Dict[str, str]] = None
limit: Optional[int] = 10
class CapabilityResponsePayload(request_id, capabilities)

Bases: object

Response payload containing discovered capabilities that match a request.

This structure enables peer-to-peer capability discovery by returning matching capabilities from agents across the network.

Parameters:
request_id: str
capabilities: List[Dict[str, Any]]
class RequestCollaborationPayload(capability_name, input_data)

Bases: object

Payload for requesting peer-to-peer collaboration on a specific capability.

This structure enables an agent to request another agent’s services based on its capabilities, forming dynamic collaborative relationships.

Parameters:
capability_name: str
input_data: Dict[str, Any]
class CollaborationResponsePayload(request_id, success, output_data=None, error_message=None)

Bases: object

Response payload from an agent providing its capability as a service.

This structure enables agents to share results after performing a requested capability, completing the peer-to-peer collaboration.

Parameters:
request_id: str
success: bool
output_data: Optional[Dict[str, Any]] = None
error_message: Optional[str] = None
class CollaborationErrorPayload(request_id, error_code, error_message)

Bases: object

Error payload for a collaboration attempt that couldn’t be completed.

This structure provides standardized error reporting in peer-to-peer collaboration scenarios.

Parameters:

  • request_id (str)

  • error_code (str)

  • error_message (str)

request_id: str
error_code: str
error_message: str
class CollaborationProtocol

Bases: BaseProtocol

Protocol that enables peer-to-peer agent collaboration and capability sharing.

This protocol facilitates dynamic discovery and utilization of capabilities across a network of independent agents. It allows agents to:

  1. Discover what other agents can do through capability queries

  2. Request help from agents with relevant capabilities

  3. Share results after task completion

Agents maintain their independence - each agent decides whether to accept collaboration requests based on its own criteria. The protocol simply standardizes the communication pattern without imposing central control.

async format_message(sender_id, receiver_id, content, sender_identity, message_type=MessageType.TEXT, metadata=None)

Format a message according to the collaboration protocol.

Return type:

Message

Parameters:
async validate_message(message)

Validate message against protocol requirements.

Return type:

bool

Parameters:

message (Message)