"""
MCP Tool Protocol Definitions
Architecture Integration:
- Design Patterns: MCP tool pattern with FastMCP integration
- Security Model: Comprehensive input validation and authorization for all tool operations
- Performance Profile: O(1) tool dispatch, O(log n) validation, async operation support
Technical Decisions:
- Protocol-based design: Enables dependency injection and comprehensive testing
- Async-first approach: All operations support concurrent execution
- Security-first validation: All inputs validated before processing
- Immutable results: All tool results are immutable for thread safety
Dependencies & Integration:
- External: fastmcp for MCP tool framework integration
- Internal: Types system for all domain objects and validation
Quality Assurance:
- Test Coverage: Protocol compliance testing with comprehensive mock implementations
- Error Handling: Structured error responses with security-focused messaging
Author: Adder_4 | Created: 2025-06-26 | Last Modified: 2025-06-26
"""
from __future__ import annotations
from abc import abstractmethod
from typing import Any, Dict, List, Optional, Protocol
from src.models.agent import (
AgentConfig,
AgentCreationRequest,
AgentCreationResult,
AgentId,
)
from src.models.communication import (
ConversationClearResult,
ConversationStartResult,
Message,
MessageResult,
)
from src.models.monitoring import HealthStatus, SessionStatusResult
from src.models.security import SecurityContext
from src.models.session import SessionCreationRequest, SessionCreationResult, SessionId
class MCPToolProtocol(Protocol):
"""
Base protocol for all MCP tool implementations with security and validation.
Contracts:
Preconditions:
- All tool parameters must be validated against security schema
- Security context must be authenticated and authorized
- Input sanitization must be performed for all user data
Postconditions:
- All operations maintain comprehensive audit trails
- Results are immutable and serializable for MCP transport
- Error conditions provide appropriate user feedback without security leaks
Invariants:
- Tool operations never bypass security boundaries
- Resource limits are enforced throughout operation lifecycle
- All state changes are atomic and consistent
"""
@abstractmethod
async def validate_inputs(
self, parameters: Dict[str, Any], security_context: SecurityContext
) -> bool:
"""
Validate tool input parameters with comprehensive security checks.
Args:
parameters: Raw tool parameters from MCP request
security_context: Security context for authorization
Returns:
True if validation successful, False otherwise
"""
...
@abstractmethod
async def execute(
self, parameters: Dict[str, Any], security_context: SecurityContext
) -> Dict[str, Any]:
"""
Execute tool operation with validated parameters.
Args:
parameters: Validated tool parameters
security_context: Security context for operation
Returns:
Tool execution result dictionary
"""
...
class CreateAgentToolProtocol(MCPToolProtocol):
"""Protocol for create_agent MCP tool implementation."""
@abstractmethod
async def create_agent(
self,
session_id: SessionId,
agent_name: str,
specialization: Optional[str],
system_prompt_suffix: str,
claude_config: Optional[AgentConfig],
security_context: SecurityContext,
) -> AgentCreationResult:
"""
Create new Claude Code agent instance with iTerm2 tab and process.
Security:
- Validates agent_name format (Agent_#)
- Ensures unique agent numbers within session
- Applies resource limits and security constraints
Process:
1. Validate session exists and has capacity
2. Generate unique agent ID and verify naming
3. Create iTerm2 tab in session context
4. Spawn Claude Code process with configuration
5. Inject ADDER+ system prompt with agent name
6. Initialize agent state and health monitoring
7. Register agent in session and persist state
"""
...
class DeleteAgentToolProtocol(MCPToolProtocol):
"""Protocol for delete_agent MCP tool implementation."""
@abstractmethod
async def delete_agent(
self, agent_name: str, force: bool, security_context: SecurityContext
) -> bool:
"""
Remove agent from system with cleanup and state persistence.
Security:
- Requires elevated permissions for force deletion
- Ensures clean shutdown and resource cleanup
- Maintains audit trail of agent lifecycle
Process:
1. Locate agent by name across all sessions
2. Send graceful shutdown signal to Claude Code process
3. Close associated iTerm2 tab
4. Clean up temporary files and resources
5. Remove agent from session state
6. Persist updated session state
7. Log deletion event with full audit trail
"""
...
class CreateSessionToolProtocol(MCPToolProtocol):
"""Protocol for create_session MCP tool implementation."""
@abstractmethod
async def create_session(
self,
root_path: str,
session_name: str,
security_level: Optional[str],
security_context: SecurityContext,
) -> SessionCreationResult:
"""
Create session tied to specific root filepath with security boundaries.
Security:
- Validates root_path accessibility and permissions
- Establishes filesystem boundaries for session
- Creates encrypted state storage
Process:
1. Validate root_path exists and is accessible
2. Create session directory structure if needed
3. Initialize Git integration and project analysis
4. Establish security context and file boundaries
5. Create session state storage with encryption
6. Initialize task file monitoring system
7. Register session in global state manager
"""
...
class GetSessionStatusToolProtocol(MCPToolProtocol):
"""Protocol for get_session_status MCP tool implementation."""
@abstractmethod
async def get_session_status(
self, session_id: Optional[SessionId], security_context: SecurityContext
) -> SessionStatusResult:
"""
Return comprehensive status of all agents in session(s).
Information Provided:
- Agent health and activity status
- Current task assignments and progress
- Resource usage and performance metrics
- iTerm2 tab status and connectivity
- Error states and recovery actions needed
Process:
1. Query all sessions (or specific session)
2. Poll agent health and process status
3. Check iTerm2 tab connectivity
4. Analyze task file states and progress
5. Calculate performance metrics
6. Aggregate status information
7. Return comprehensive status report
"""
...
class DeleteSessionToolProtocol(MCPToolProtocol):
"""Protocol for delete_session MCP tool implementation."""
@abstractmethod
async def delete_session(
self,
session_id: SessionId,
cleanup_agents: bool,
preserve_work: bool,
security_context: SecurityContext,
) -> bool:
"""
Remove entire session and optionally all associated agents.
Security:
- Requires confirmation for destructive operations
- Preserves work products by default
- Maintains audit trail of session lifecycle
Process:
1. Validate session exists and permissions
2. Optionally shutdown all session agents gracefully
3. Close all associated iTerm2 tabs
4. Clean up session-specific resources
5. Preserve or clean work files based on flags
6. Remove session from global state
7. Log session destruction with full audit
"""
...
class SendMessageToAgentToolProtocol(MCPToolProtocol):
"""Protocol for send_message_to_agent MCP tool implementation."""
@abstractmethod
async def send_message_to_agent(
self,
agent_name: str,
message: str,
prepend_adder: bool,
wait_for_response: bool,
security_context: SecurityContext,
) -> MessageResult:
"""
Send message to specific agent with automatic ADDER+ prepending.
Message Processing:
- Automatically prepends ADDER+ system prompt
- Injects agent name into system prompt
- Handles message formatting and escaping
- Manages iTerm2 text injection
Process:
1. Locate agent and validate it's active (use `/debug` if issues)
2. Construct full message with ADDER+ prompt
3. Inject agent name into system prompt
4. Send formatted message via iTerm2 text injection to session directory
5. Monitor agent response if requested (use `/logs` for debugging)
6. Log interaction with full message content
7. Return delivery confirmation and response
"""
...
class ClearAgentConversationToolProtocol(MCPToolProtocol):
"""Protocol for clear_agent_conversation MCP tool implementation."""
@abstractmethod
async def clear_agent_conversation(
self, agent_name: str, preserve_state: bool, security_context: SecurityContext
) -> ConversationClearResult:
"""
Close current iTerm2 tab for agent (clears conversation).
State Management:
- Preserves agent configuration and task assignments
- Clears conversation history but maintains context
- Gracefully shuts down Claude Code process
Process:
1. Locate agent and verify it's active
2. Send shutdown signal to Claude Code process
3. Close associated iTerm2 tab
4. Clear conversation history from agent state
5. Preserve task context and configuration
6. Update agent status to IDLE
7. Log conversation clear event
"""
...
class StartNewAgentConversationToolProtocol(MCPToolProtocol):
"""Protocol for start_new_agent_conversation MCP tool implementation."""
@abstractmethod
async def start_new_agent_conversation(
self, agent_name: str, restore_context: bool, security_context: SecurityContext
) -> ConversationStartResult:
"""
Open new iTerm2 tab for agent (starts fresh conversation).
Context Management:
- Restores agent configuration and specialization
- Optionally restores recent task context
- Reinitializes Claude Code with preserved settings
Process:
1. Locate agent and verify it exists
2. Create new iTerm2 tab in session context
3. Spawn new Claude Code process with preserved config
4. Inject ADDER+ prompt with agent context
5. Optionally restore recent conversation context
6. Update agent state with new process/tab IDs
7. Log new conversation start event
"""
...
class MCPToolRegistryProtocol(Protocol):
"""Protocol for MCP tool registration and management."""
@abstractmethod
async def register_tool(
self,
tool_name: str,
tool_implementation: MCPToolProtocol,
schema: Dict[str, Any],
) -> bool:
"""
Register MCP tool with FastMCP server.
Args:
tool_name: Unique tool identifier
tool_implementation: Tool protocol implementation
schema: JSON schema for tool parameters
Returns:
True if registration successful, False otherwise
"""
...
@abstractmethod
async def unregister_tool(self, tool_name: str) -> bool:
"""
Unregister MCP tool from server.
Args:
tool_name: Tool identifier to remove
Returns:
True if unregistration successful, False otherwise
"""
...
@abstractmethod
async def list_tools(self) -> List[str]:
"""
List all registered MCP tool names.
Returns:
List of registered tool names
"""
...
@abstractmethod
async def get_tool_schema(self, tool_name: str) -> Optional[Dict[str, Any]]:
"""
Get JSON schema for specific tool.
Args:
tool_name: Tool identifier
Returns:
Tool schema dictionary if tool exists, None otherwise
"""
...
class AsyncEventHandlerProtocol(Protocol):
"""Protocol for async event handling with comprehensive monitoring."""
@abstractmethod
async def start_monitoring(self) -> None:
"""Start async event monitoring loop."""
...
@abstractmethod
async def stop_monitoring(self) -> None:
"""Stop async event monitoring with cleanup."""
...
@abstractmethod
async def emit_event(
self,
event_type: str,
event_data: Dict[str, Any],
security_context: SecurityContext,
) -> None:
"""
Emit event to monitoring system.
Args:
event_type: Type of event (agent_created, session_deleted, etc.)
event_data: Event-specific data payload
security_context: Security context for audit trail
"""
...
@abstractmethod
async def subscribe_to_events(
self,
event_types: List[str],
callback: callable,
security_context: SecurityContext,
) -> str:
"""
Subscribe to specific event types.
Args:
event_types: List of event types to monitor
callback: Async callback function for events
security_context: Security context for authorization
Returns:
Subscription ID for later unsubscribing
"""
...
@abstractmethod
async def unsubscribe_from_events(
self, subscription_id: str, security_context: SecurityContext
) -> bool:
"""
Unsubscribe from event monitoring.
Args:
subscription_id: Subscription ID to cancel
security_context: Security context for authorization
Returns:
True if unsubscription successful, False otherwise
"""
...
