my-pal-mcp-server/docs/architecture/components.md

System Components & Interactions

Component Architecture

The Gemini MCP Server is built on a modular component architecture that enables sophisticated AI collaboration patterns while maintaining security and performance.

Core Components

1. MCP Protocol Engine

Location: server.py:45-120 Purpose: Central communication hub implementing Model Context Protocol specification

Key Responsibilities:

• Protocol Compliance: Implements MCP v1.0 specification for Claude integration
• Message Routing: Dispatches requests to appropriate tool handlers
• Error Handling: Graceful degradation and error response formatting
• Lifecycle Management: Server startup, shutdown, and resource cleanup

Implementation Details:

# server.py:67
@server.list_tools()
async def list_tools() -> list[types.Tool]:
    """Dynamic tool discovery and registration"""
    return [tool.get_schema() for tool in REGISTERED_TOOLS]

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    """Tool execution with error handling and response formatting"""

Dependencies:

• mcp library for protocol implementation
• asyncio for concurrent request processing
• Tool registry for dynamic handler discovery

2. Tool Architecture System

Location: tools/ directory Purpose: Modular plugin system for specialized AI capabilities

BaseTool Abstract Class (tools/base.py:25)

Interface Contract:

class BaseTool(ABC):
    @abstractmethod
    async def execute(self, request: dict) -> ToolOutput:
        """Core tool execution logic"""
        
    @abstractmethod
    def get_schema(self) -> types.Tool:
        """MCP tool schema definition"""
        
    def _format_response(self, content: str, metadata: dict) -> ToolOutput:
        """Standardized response formatting"""

Individual Tool Components

Chat Tool (tools/chat.py:30)

• Purpose: Quick questions and general collaboration
• Thinking Mode: Default 'medium' (8192 tokens)
• Use Cases: Brainstorming, simple explanations, immediate answers

ThinkDeep Tool (tools/thinkdeep.py:45)

• Purpose: Complex analysis and strategic planning
• Thinking Mode: Default 'high' (16384 tokens)
• Use Cases: Architecture decisions, design exploration, comprehensive analysis

CodeReview Tool (tools/codereview.py:60)

• Purpose: Code quality and security analysis
• Thinking Mode: Default 'medium' (8192 tokens)
• Use Cases: Bug detection, security audits, quality validation

Analyze Tool (tools/analyze.py:75)

• Purpose: Codebase exploration and understanding
• Thinking Mode: Variable based on scope
• Use Cases: Dependency analysis, pattern detection, system comprehension

Debug Tool (tools/debug.py:90)

• Purpose: Error investigation and root cause analysis
• Thinking Mode: Default 'medium' (8192 tokens)
• Use Cases: Stack trace analysis, bug diagnosis, performance issues

Precommit Tool (tools/precommit.py:105)

• Purpose: Automated quality gates and validation
• Thinking Mode: Default 'medium' (8192 tokens)
• Use Cases: Pre-commit validation, change analysis, quality assurance

3. Security Engine

Location: utils/file_utils.py:45-120 Purpose: Multi-layer security validation and enforcement

Security Components

Path Validation System:

# utils/file_utils.py:67
def validate_file_path(file_path: str) -> bool:
    """Multi-layer path security validation"""
    # 1. Dangerous path detection
    dangerous_patterns = ['../', '~/', '/etc/', '/var/', '/usr/']
    if any(pattern in file_path for pattern in dangerous_patterns):
        return False
    
    # 2. Absolute path requirement
    if not os.path.isabs(file_path):
        return False
    
    # 3. Sandbox boundary enforcement
    return file_path.startswith(PROJECT_ROOT)

Docker Path Translation:

# utils/file_utils.py:89
def translate_docker_path(host_path: str) -> str:
    """Convert host paths to container paths for Docker environment"""
    if host_path.startswith(WORKSPACE_ROOT):
        return host_path.replace(WORKSPACE_ROOT, '/workspace', 1)
    return host_path

Security Layers:

1. Input Sanitization: Path cleaning and normalization
2. Pattern Matching: Dangerous path detection and blocking
3. Boundary Enforcement: PROJECT_ROOT containment validation
4. Container Translation: Safe host-to-container path mapping

4. Conversation Memory System

Location: utils/conversation_memory.py:30-150 Purpose: Cross-session context preservation and threading

Memory Components

Thread Context Management:

# utils/conversation_memory.py:45
class ThreadContext:
    thread_id: str
    tool_history: List[ToolExecution]
    conversation_files: Set[str]
    context_tokens: int
    created_at: datetime
    last_accessed: datetime

Redis Integration:

# utils/conversation_memory.py:78
class ConversationMemory:
    def __init__(self, redis_url: str):
        self.redis = redis.from_url(redis_url)
    
    async def store_thread(self, context: ThreadContext) -> None:
        """Persist conversation thread to Redis"""
    
    async def retrieve_thread(self, thread_id: str) -> Optional[ThreadContext]:
        """Reconstruct conversation from storage"""
    
    async def cleanup_expired_threads(self) -> int:
        """Remove old conversations to manage memory"""

Memory Features:

• Thread Persistence: UUID-based conversation storage
• Context Reconstruction: Full conversation history retrieval
• File Deduplication: Efficient storage of repeated file references
• Automatic Cleanup: Time-based thread expiration

5. File Processing Pipeline

Location: utils/file_utils.py:120-200 Purpose: Token-aware file reading and content optimization

Processing Components

Priority System:

# utils/file_utils.py:134
FILE_PRIORITIES = {
    '.py': 1,    # Python source code (highest priority)
    '.js': 1,    # JavaScript source
    '.ts': 1,    # TypeScript source
    '.md': 2,    # Documentation
    '.txt': 3,   # Text files
    '.log': 4,   # Log files (lowest priority)
}

Token Management:

# utils/file_utils.py:156
def read_file_with_token_limit(file_path: str, max_tokens: int) -> str:
    """Read file content with token budget enforcement"""
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            content = f.read()
        
        # Token estimation and truncation
        estimated_tokens = len(content) // 4  # Rough estimation
        if estimated_tokens > max_tokens:
            # Truncate with preservation of structure
            content = content[:max_tokens * 4]
        
        return format_file_content(content, file_path)
    except Exception as e:
        return f"Error reading {file_path}: {str(e)}"

Content Formatting:

• Line Numbers: Added for precise code references
• Error Handling: Graceful failure with informative messages
• Structure Preservation: Maintains code formatting and indentation

6. Gemini API Integration

Location: tools/models.py:25-80 Purpose: Standardized interface to Google's Gemini models

Integration Components

API Client:

# tools/models.py:34
class GeminiClient:
    def __init__(self, api_key: str, model: str = "gemini-2.0-flash-thinking-exp"):
        self.client = genai.GenerativeModel(model)
        self.api_key = api_key
    
    async def generate_response(self, 
                              prompt: str, 
                              thinking_mode: str = 'medium',
                              files: List[str] = None) -> str:
        """Generate response with thinking mode and file context"""

Model Configuration:

# config.py:24
GEMINI_MODEL = os.getenv('GEMINI_MODEL', 'gemini-2.0-flash-thinking-exp')
MAX_CONTEXT_TOKENS = int(os.getenv('MAX_CONTEXT_TOKENS', '1000000'))

Thinking Mode Management:

# tools/models.py:67
THINKING_MODE_TOKENS = {
    'minimal': 128,
    'low': 2048,
    'medium': 8192,
    'high': 16384,
    'max': 32768
}

Component Interactions

1. Request Processing Flow

Claude Request
    ↓
MCP Protocol Engine (server.py:67)
    ↓ (validate & route)
Tool Selection & Loading
    ↓
Security Validation (utils/file_utils.py:67)
    ↓ (if files involved)
File Processing Pipeline (utils/file_utils.py:134)
    ↓
Conversation Context Loading (utils/conversation_memory.py:78)
    ↓ (if continuation_id provided)
Gemini API Integration (tools/models.py:34)
    ↓
Response Processing & Formatting
    ↓
Conversation Storage (utils/conversation_memory.py:78)
    ↓
MCP Response to Claude

2. Security Integration Points

Pre-Tool Execution:

• Path validation before any file operations
• Sandbox boundary enforcement
• Docker path translation for container environments

During Tool Execution:

• Token budget enforcement to prevent memory exhaustion
• File access logging and monitoring
• Error containment and graceful degradation

Post-Tool Execution:

• Response sanitization
• Conversation storage with access controls
• Resource cleanup and memory management

3. Memory System Integration

Thread Creation:

# New conversation
thread_id = str(uuid.uuid4())
context = ThreadContext(thread_id=thread_id, ...)
await memory.store_thread(context)

Thread Continuation:

# Continuing conversation
if continuation_id:
    context = await memory.retrieve_thread(continuation_id)
    # Merge new request with existing context

Cross-Tool Communication:

# Tool A stores findings
await memory.add_tool_execution(thread_id, tool_execution)

# Tool B retrieves context
context = await memory.retrieve_thread(thread_id)
previous_findings = context.get_tool_outputs('analyze')

Configuration & Dependencies

Environment Configuration

Required Settings (config.py):

GEMINI_API_KEY = os.getenv('GEMINI_API_KEY')  # Required
GEMINI_MODEL = os.getenv('GEMINI_MODEL', 'gemini-2.0-flash-thinking-exp')
PROJECT_ROOT = os.getenv('PROJECT_ROOT', '/workspace')
REDIS_URL = os.getenv('REDIS_URL', 'redis://localhost:6379')
MAX_CONTEXT_TOKENS = int(os.getenv('MAX_CONTEXT_TOKENS', '1000000'))

Component Dependencies

Core Dependencies:

• mcp: MCP protocol implementation
• google-generativeai: Gemini API client
• redis: Conversation persistence
• asyncio: Concurrent processing

Security Dependencies:

• pathlib: Path manipulation and validation
• os: File system operations and environment access

Tool Dependencies:

• pydantic: Data validation and serialization
• typing: Type hints and contract definition

Extension Architecture

Adding New Components

1. Tool Components: Inherit from BaseTool and implement required interface
2. Security Components: Extend validation chain in file_utils.py
3. Memory Components: Add new storage backends via interface abstraction
4. Processing Components: Extend file pipeline with new content types

Integration Patterns

• Plugin Architecture: Dynamic discovery and registration
• Interface Segregation: Clear contracts between components
• Dependency Injection: Configuration-driven component assembly
• Error Boundaries: Isolated failure handling per component

---

This component architecture provides a robust foundation for AI collaboration while maintaining security, performance, and extensibility requirements.