From 92c0d5d7701a196bcec97393be367e4ae89c4fa6 Mon Sep 17 00:00:00 2001 From: Jeroen Schweitzer Date: Thu, 11 Dec 2025 21:26:52 +0100 Subject: [PATCH] feat(phase3): add agent communication protocol MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - AgentRequest/AgentResponse for standardized inter-agent communication - DelegationIntent for routing tasks to expert agents - CoordinationResult for aggregated multi-agent results - DelegationReason enum (domain expertise, tool access, etc.) - Error types: AgentError, AgentTimeoutError, AgentUnavailableError 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 --- src/agents/protocol.py | 201 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 201 insertions(+) create mode 100644 src/agents/protocol.py diff --git a/src/agents/protocol.py b/src/agents/protocol.py new file mode 100644 index 0000000..70dc00b --- /dev/null +++ b/src/agents/protocol.py @@ -0,0 +1,201 @@ +""" +Agent communication protocol for multi-agent coordination. + +Defines standardized request/response formats for communication between: +- Steward (request analysis) → Tatlock (coordination) +- Tatlock (coordination) → Expert agents (Librarian, Developer, etc.) +""" +from enum import Enum +from typing import Any, Optional + +from pydantic import BaseModel, Field + + +class DelegationReason(str, Enum): + """Why a task is being delegated to an expert agent.""" + DOMAIN_EXPERTISE = "domain_expertise" # Expert has specialized knowledge + TOOL_ACCESS = "tool_access" # Expert has required tools + RESOURCE_EFFICIENCY = "resource_efficiency" # Better handled by specialist + USER_PREFERENCE = "user_preference" # User requested specific agent + + +class TaskComplexity(str, Enum): + """Complexity estimate for task execution.""" + SIMPLE = "simple" # Single tool call, fast + MODERATE = "moderate" # Multiple steps, moderate time + COMPLEX = "complex" # Multi-agent, significant processing + + +class AgentRequest(BaseModel): + """ + Request to an expert agent. + + Contains everything the agent needs to execute a task, + including context from the conversation and delegation intent. + """ + task: str = Field( + ..., + description="Clear description of what the agent should do" + ) + context: str = Field( + default="", + description="Relevant context from conversation history" + ) + constraints: list[str] = Field( + default_factory=list, + description="Any constraints or requirements for the task" + ) + delegation_reason: DelegationReason = Field( + default=DelegationReason.DOMAIN_EXPERTISE, + description="Why this task was delegated to this agent" + ) + user_id: str = Field( + default="default", + description="User identifier for multi-tenant operations" + ) + max_tokens: Optional[int] = Field( + default=None, + description="Optional token limit for response" + ) + timeout_seconds: Optional[int] = Field( + default=60, + description="Maximum time for task completion" + ) + + +class ToolCallRecord(BaseModel): + """Record of a tool call made during execution.""" + tool_name: str + arguments: dict[str, Any] + result: str + duration_ms: int + + +class AgentResponse(BaseModel): + """ + Response from an expert agent. + + Contains the result, reasoning, and metadata about execution. + """ + success: bool = Field( + ..., + description="Whether the task completed successfully" + ) + result: str = Field( + ..., + description="The main output/answer from the agent" + ) + reasoning: str = Field( + default="", + description="Agent's reasoning process (for transparency)" + ) + tool_calls: list[ToolCallRecord] = Field( + default_factory=list, + description="Tools called during execution" + ) + confidence: float = Field( + default=1.0, + ge=0.0, + le=1.0, + description="Agent's confidence in the result (0.0-1.0)" + ) + sources: list[str] = Field( + default_factory=list, + description="Sources or references used" + ) + error_message: Optional[str] = Field( + default=None, + description="Error details if success=False" + ) + duration_ms: int = Field( + default=0, + description="Total execution time in milliseconds" + ) + + +class DelegationIntent(BaseModel): + """ + Intent to delegate a task to an expert agent. + + Created by Tatlock when deciding to delegate, based on + Steward's recommendations. + """ + target_agent: str = Field( + ..., + description="Name of the expert agent to delegate to" + ) + task: str = Field( + ..., + description="Task description for the agent" + ) + reason: DelegationReason = Field( + default=DelegationReason.DOMAIN_EXPERTISE, + description="Why delegating to this agent" + ) + expected_outcome: str = Field( + default="", + description="What we expect the agent to provide" + ) + priority: int = Field( + default=1, + ge=1, + le=10, + description="Priority (1=highest, 10=lowest)" + ) + depends_on: list[str] = Field( + default_factory=list, + description="Other delegation IDs this depends on (for sequencing)" + ) + + +class CoordinationResult(BaseModel): + """ + Result of multi-agent coordination. + + Aggregates results from multiple expert agents into + a single coherent response. + """ + final_response: str = Field( + ..., + description="Synthesized response from all agents" + ) + agent_responses: dict[str, AgentResponse] = Field( + default_factory=dict, + description="Individual responses keyed by agent name" + ) + delegation_intents: list[DelegationIntent] = Field( + default_factory=list, + description="All delegations that were executed" + ) + total_duration_ms: int = Field( + default=0, + description="Total coordination time" + ) + agents_consulted: list[str] = Field( + default_factory=list, + description="Names of agents that contributed" + ) + + +class AgentError(Exception): + """Base exception for agent errors.""" + + def __init__(self, message: str, agent_name: str = "unknown"): + self.message = message + self.agent_name = agent_name + super().__init__(f"[{agent_name}] {message}") + + +class AgentTimeoutError(AgentError): + """Agent execution timed out.""" + pass + + +class AgentUnavailableError(AgentError): + """Agent is not available or registered.""" + pass + + +class DelegationError(AgentError): + """Error during task delegation.""" + pass