Handoff Context Serialization¶

from dataclasses import dataclass, field
from typing import Any

@dataclass
class HandoffContext:
    """Context data for an agent handoff.

    Attributes:
        conversation_history: List of messages in the conversation history.
            Each message is a dict with 'role', 'content', and optional metadata.
        tool_state: Current state of tools, including active tool calls,
            cached results, and tool configurations.
        metadata: Additional metadata like timestamps, agent capabilities,
            resource allocations, etc.
    """
    conversation_history: list[dict[str, Any]] = field(default_factory=list)
    tool_state: dict[str, Any] = field(default_factory=dict)
    metadata: dict[str, Any] = field(default_factory=dict)

    def to_dict(self) -> dict[str, Any]:
        """Convert the context to a dictionary."""
        return asdict(self)

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> HandoffContext:
        """Create a HandoffContext from a dictionary."""
        return cls(
            conversation_history=data.get("conversation_history", []),
            tool_state=data.get("tool_state", {}),
            metadata=data.get("metadata", {})
        )

conversation_history = [
    {
        "role": "user",
        "content": "Analyze the sales data for Q3",
        "timestamp": "2025-01-15T10:30:00Z"
    },
    {
        "role": "assistant",
        "content": "I'll analyze the Q3 sales data for you.",
        "timestamp": "2025-01-15T10:30:01Z"
    },
    {
        "role": "tool",
        "name": "query_database",
        "tool_call_id": "call_abc123",
        "content": '{"total_sales": 1500000, "growth": 0.12}',
        "timestamp": "2025-01-15T10:30:05Z"
    }
]

tool_state = {
    "active_calls": [
        {"call_id": "call_xyz789", "tool": "long_running_query", "started_at": "..."}
    ],
    "cached_results": {
        "call_abc123": {"total_sales": 1500000, "growth": 0.12}
    },
    "configurations": {
        "database": {"connection_string": "...", "timeout_ms": 30000}
    }
}

def serialize_context(context: HandoffContext) -> bytes:
    """Serialize HandoffContext to compact JSON bytes.

    Uses UTF-8 encoding with compact separators (',', ':') to minimize size.

    Args:
        context: The HandoffContext to serialize

    Returns:
        JSON-encoded bytes representation of the context

    Raises:
        ValueError: If the context cannot be serialized (e.g., non-JSON-serializable values)
    """

from sw4rm.handoff.context import HandoffContext, serialize_context

context = HandoffContext(
    conversation_history=[
        {"role": "user", "content": "Hello"},
        {"role": "assistant", "content": "Hi there!"}
    ],
    tool_state={},
    metadata={"agent_id": "agent-1"}
)

data = serialize_context(context)
# b'{"conversation_history":[{"role":"user","content":"Hello"},...],...}'

def deserialize_context(data: bytes) -> HandoffContext:
    """Deserialize bytes to a HandoffContext.

    Args:
        data: JSON-encoded bytes representation of a context

    Returns:
        HandoffContext instance

    Raises:
        ValueError: If the data cannot be deserialized
            - Invalid JSON
            - Invalid UTF-8 encoding
            - Missing required fields
    """

from sw4rm.handoff.context import deserialize_context

data = b'{"conversation_history":[],"tool_state":{},"metadata":{}}'
context = deserialize_context(data)

from dataclasses import dataclass, field
from typing import Optional, Any

@dataclass
class HandoffRequest:
    """Request to hand off execution to another agent.

    Attributes:
        from_agent: ID of the agent initiating the handoff
        to_agent: ID of the agent that should receive the handoff
        reason: Human-readable explanation of why the handoff is needed
        context_snapshot: Serialized HandoffContext (from serialize_context)
        preserve_history: Whether to preserve conversation history (default: True)
        capabilities_required: List of capabilities the target agent must have
        metadata: Additional metadata for the handoff
    """
    from_agent: str
    to_agent: str
    reason: str
    context_snapshot: Optional[bytes] = None
    preserve_history: bool = True
    capabilities_required: list[str] = field(default_factory=list)
    metadata: dict[str, Any] = field(default_factory=dict)

from dataclasses import dataclass, field
from typing import Optional, Any
from enum import Enum

class HandoffStatus(Enum):
    """Status of a handoff request."""
    PENDING = "PENDING"
    ACCEPTED = "ACCEPTED"
    REJECTED = "REJECTED"
    COMPLETED = "COMPLETED"

@dataclass
class HandoffResponse:
    """Response to a handoff request.

    Attributes:
        accepted: Whether the handoff was accepted
        handoff_id: Unique identifier for the handoff transaction
        rejection_reason: Explanation if the handoff was rejected
        status: Current status of the handoff
        metadata: Additional metadata about the handoff
    """
    accepted: bool
    handoff_id: str
    rejection_reason: Optional[str] = None
    status: HandoffStatus = HandoffStatus.PENDING
    metadata: dict[str, Any] = field(default_factory=dict)

def on_handoff_request(self, request: HandoffRequest) -> HandoffResponse:
    """Called when another agent requests handoff to this agent.

    Override to implement acceptance logic based on:
    - capabilities_required matching your agent's capabilities
    - Current agent load/state (are you available?)
    - Policy checks (is this handoff allowed?)

    Args:
        request: HandoffRequest from the requesting agent

    Returns:
        HandoffResponse indicating acceptance or rejection
    """

class SpecialistAgent(Agent):
    CAPABILITIES = ["code_review", "security_analysis"]

    def on_handoff_request(self, request: HandoffRequest) -> HandoffResponse:
        # Check capabilities
        for cap in request.capabilities_required:
            if cap not in self.CAPABILITIES:
                return HandoffResponse(
                    accepted=False,
                    handoff_id=str(uuid.uuid4()),
                    rejection_reason=f"Missing capability: {cap}",
                    status=HandoffStatus.REJECTED
                )

        # Check availability
        if self.state != AgentState.RUNNABLE:
            return HandoffResponse(
                accepted=False,
                handoff_id=str(uuid.uuid4()),
                rejection_reason=f"Agent busy: {self.state_name}",
                status=HandoffStatus.REJECTED
            )

        # Accept handoff
        return HandoffResponse(
            accepted=True,
            handoff_id=str(uuid.uuid4()),
            status=HandoffStatus.ACCEPTED,
            metadata={"capabilities": self.CAPABILITIES}
        )

def on_handoff_received(self, context: HandoffContext) -> None:
    """Called after handoff is accepted and context is received.

    Override to restore state from context:
    - Conversation history for continuity
    - Tool state for resuming operations
    - Session metadata for context awareness

    Args:
        context: HandoffContext with conversation history, tool state, etc.
    """

class SpecialistAgent(Agent):
    def on_handoff_received(self, context: HandoffContext) -> None:
        # Restore conversation history
        self.conversation_history = context.conversation_history
        self.logger.info(f"Received {len(self.conversation_history)} history entries")

        # Restore tool state
        if context.tool_state.get("active_calls"):
            for call in context.tool_state["active_calls"]:
                self._resume_tool_call(call)

        # Apply metadata
        self.session_metadata = context.metadata
        self.logger.info(f"Handoff received from: {context.metadata.get('agent_id')}")

from sw4rm.handoff import HandoffRequest, HandoffContext
from sw4rm.handoff.context import serialize_context

# Source agent initiates handoff
context = HandoffContext(
    conversation_history=[
        {"role": "user", "content": "Review this code for security issues"},
        {"role": "assistant", "content": "I'll need to hand this off to a specialist."}
    ],
    tool_state={},
    metadata={"original_request_id": "req-123"}
)

request = HandoffRequest(
    from_agent="general-agent-1",
    to_agent="security-specialist-1",
    reason="Task requires specialized security analysis capabilities",
    context_snapshot=serialize_context(context),
    preserve_history=True,
    capabilities_required=["security_analysis"],
    metadata={"urgency": "high"}
)

# Send request via HandoffClient
response = handoff_client.request_handoff(request)

if response.accepted:
    print(f"Handoff accepted: {response.handoff_id}")
else:
    print(f"Handoff rejected: {response.rejection_reason}")

class CapabilityRouter:
    """Route handoffs based on required capabilities."""

    def __init__(self, registry_client):
        self.registry = registry_client

    def find_capable_agent(self, capabilities: list[str]) -> Optional[str]:
        """Find an agent with all required capabilities."""
        agents = self.registry.list_agents()

        for agent in agents:
            agent_caps = set(agent.capabilities)
            if all(cap in agent_caps for cap in capabilities):
                if agent.state == "RUNNABLE":
                    return agent.agent_id

        return None

    def route_handoff(self, request: HandoffRequest) -> HandoffResponse:
        """Route a handoff to an appropriate agent."""
        if not request.to_agent:
            # Find suitable agent
            target = self.find_capable_agent(request.capabilities_required)
            if not target:
                return HandoffResponse(
                    accepted=False,
                    handoff_id=str(uuid.uuid4()),
                    rejection_reason="No capable agent available",
                    status=HandoffStatus.REJECTED
                )
            request.to_agent = target

        return self.handoff_client.request_handoff(request)

def initiate_handoff_with_fallback(
    from_agent: str,
    context: HandoffContext,
    preferred_agents: list[str],
    capabilities: list[str]
) -> HandoffResponse:
    """Try handoff to preferred agents with fallback."""

    for agent_id in preferred_agents:
        request = HandoffRequest(
            from_agent=from_agent,
            to_agent=agent_id,
            reason="Task delegation",
            context_snapshot=serialize_context(context),
            capabilities_required=capabilities
        )

        response = handoff_client.request_handoff(request)

        if response.accepted:
            return response

        print(f"Agent {agent_id} rejected: {response.rejection_reason}")

    # All preferred agents rejected
    return HandoffResponse(
        accepted=False,
        handoff_id=str(uuid.uuid4()),
        rejection_reason="All preferred agents unavailable",
        status=HandoffStatus.REJECTED
    )

def prepare_handoff_context(
    full_history: list[dict],
    tool_state: dict,
    max_history: int = 50
) -> HandoffContext:
    """Prepare a size-optimized handoff context."""

    # Trim history to recent messages
    trimmed_history = full_history[-max_history:]

    # Remove completed tool results
    active_tool_state = {
        "active_calls": tool_state.get("active_calls", []),
        "pending_approvals": tool_state.get("pending_approvals", []),
        # Omit cached_results to save space
    }

    return HandoffContext(
        conversation_history=trimmed_history,
        tool_state=active_tool_state,
        metadata={
            "original_history_length": len(full_history),
            "trimmed": len(full_history) > max_history
        }
    )