Tutorial What Is An MCP Server - 80-20-Human-In-The-Loop/Community GitHub Wiki

Understanding MCP Servers

MCP (Model Context Protocol) standardizes how LLMs interact with external systems through a client-server architecture.

Table of Contents

  1. Introduction
  2. The Three Powers of MCP
  3. How Your AI Uses MCP
  4. The 80-20 Balance
  5. Real-World Examples
  6. Why This Matters

Introduction

Large Language Models excel at natural language processing but lack direct access to external systems. They can discuss file operations, describe computations, and explain database queries - but cannot execute these operations directly.

MCP (Model Context Protocol) bridges this gap.

MCP provides a standardized protocol for LLMs to interact with external tools and data sources through well-defined interfaces. The protocol establishes secure, controlled channels for AI systems to access real-world functionality while maintaining clear boundaries and permissions.

Target Audiences

  • Software Engineers: Enable AI to analyze codebases, execute tests, and interact with development tools
  • Data Engineers: Connect AI to databases, data pipelines, and analytics platforms
  • System Architects: Design secure, scalable AI integrations with existing infrastructure

Core MCP Capabilities

MCP servers expose three primary capability types through a JSON-RPC 2.0 interface:

1. Tools: Executable Functions

Definition: Functions that LLMs can invoke with typed parameters to perform operations.

Protocol Structure: Tools are exposed as callable methods with defined input/output schemas.

Common Implementations:

  • Database operations (queries, updates, transactions)
  • File system operations (read, write, delete)
  • API integrations (REST, GraphQL, gRPC)
  • Computational tasks (data processing, analysis)
  • System commands (process management, monitoring)

Implementation Example:

@server.tool()
async def execute_query(
    query: str,
    params: Optional[Dict[str, Any]] = None,
    timeout: int = 30
) -> Dict[str, Any]:
    """Execute parameterized database query with connection pooling.
    
    Returns:
        Dict containing result set, metadata, and execution statistics
    """
    async with get_db_connection() as conn:
        result = await conn.execute(query, params, timeout=timeout)
        return {
            "data": result.rows,
            "metadata": result.metadata,
            "execution_time_ms": result.execution_time
        }

Execution Flow: The LLM sends a JSON-RPC request with method name and parameters. The server validates inputs, executes the function, and returns structured results.

2. Resources: Structured Data Access

Definition: Read-only data sources that provide contextual information to LLMs.

Protocol Structure: Resources are exposed with URIs and MIME types, supporting both synchronous and streaming access patterns.

Resource Types:

  • Static files (documentation, configurations, schemas)
  • Dynamic data (metrics, logs, state information)
  • Structured datasets (databases, APIs, knowledge graphs)
  • Binary content (images, documents, archives)

Implementation Example:

@server.resource(
    uri="project://docs/{path:path}",
    mime_type="text/markdown"
)
async def get_documentation(path: str) -> Resource:
    """Provide access to project documentation with caching.
    
    Returns:
        Resource object with content, metadata, and refresh policy
    """
    content = await load_cached_doc(path)
    return Resource(
        content=content,
        metadata={
            "last_modified": get_modification_time(path),
            "cache_control": "max-age=3600",
            "etag": calculate_etag(content)
        }
    )

Access Patterns: Resources support HTTP-style caching semantics, partial content requests, and content negotiation for optimal performance.

3. Prompts: Workflow Templates

Definition: Structured templates that define reusable interaction patterns and workflows.

Protocol Structure: Prompts include parameters, context requirements, and output specifications.

Use Cases:

  • Standardized analysis procedures
  • Multi-step workflows with decision points
  • Domain-specific interaction patterns
  • Compliance and audit templates
  • Quality assurance checklists

Implementation Example:

@server.prompt(
    name="code_review",
    parameters=[
        Parameter(name="file_path", type="string", required=True),
        Parameter(name="review_level", type="enum", values=["basic", "thorough", "security"])
    ]
)
def generate_code_review_prompt(file_path: str, review_level: str = "basic") -> Prompt:
    """Generate structured code review workflow.
    
    Returns:
        Prompt with review criteria, context requirements, and output format
    """
    return Prompt(
        messages=[
            {"role": "system", "content": get_review_criteria(review_level)},
            {"role": "user", "content": f"Review: {file_path}"}
        ],
        context_required=[f"file://{file_path}", "git://history"],
        output_schema=review_output_schema(review_level)
    )

Execution Model: Prompts define the interaction contract between client and server, ensuring consistent behavior across different LLM implementations.

Client-Server Architecture

Protocol Communication

MCP uses JSON-RPC 2.0 over stdio or HTTP(S) transports:

// Client request
{
  "jsonrpc": "2.0",
  "id": "req-001",
  "method": "tools/call",
  "params": {
    "name": "execute_query",
    "arguments": {
      "query": "SELECT * FROM users WHERE active = ?",
      "params": {"active": true}
    }
  }
}

// Server response
{
  "jsonrpc": "2.0",
  "id": "req-001",
  "result": {
    "data": [...],
    "metadata": {"row_count": 42, "execution_time_ms": 23}
  }
}

Connection Lifecycle

  1. Initialization: Client establishes connection and negotiates capabilities
  2. Discovery: Client queries available tools, resources, and prompts
  3. Interaction: Client invokes capabilities as needed
  4. Cleanup: Graceful shutdown with resource cleanup

Client Implementations

Claude Code: Native MCP support with configuration in claude_code_config.json

VS Code Extensions: Integration through Language Server Protocol bridge

Custom Applications: Direct implementation using MCP SDKs (Python, TypeScript, Go)

Server Deployment

Servers can be deployed as:

  • Local processes (development, personal tools)
  • Network services (team collaboration, shared resources)
  • Containerized applications (scalable production deployments)
  • Serverless functions (event-driven architectures)

The 80-20 Human-in-the-Loop Integration

MCP servers implement the 80-20 philosophy through explicit capability boundaries:

Automated Operations (80%)

  • Pattern recognition across large codebases
  • Repetitive task execution with defined parameters
  • Data aggregation and preliminary analysis
  • Standard workflow execution
  • Routine maintenance operations

Human-Controlled Decisions (20%)

  • Capability authorization and permission management
  • Critical operation approval workflows
  • Architecture and design decisions
  • Security policy configuration
  • Business logic validation

Permission Model

class MCPPermissions:
    """Define capability access controls."""
    
    def __init__(self):
        self.policies = {
            "read_operations": {"auto_approve": True},
            "write_operations": {"auto_approve": False, "requires": "user_confirmation"},
            "delete_operations": {"auto_approve": False, "requires": "admin_approval"},
            "system_operations": {"auto_approve": False, "requires": "explicit_grant"}
        }
    
    @server.tool()
    async def execute_operation(self, operation: str, params: Dict) -> Dict:
        """Execute operation with permission checks."""
        policy = self.get_policy(operation)
        
        if not policy["auto_approve"]:
            approval = await request_approval(
                operation=operation,
                params=params,
                requirement=policy["requires"]
            )
            if not approval.granted:
                return {"status": "denied", "reason": approval.reason}
        
        return await perform_operation(operation, params)

The permission model ensures critical operations remain under human control while routine tasks execute efficiently.

Implementation Examples

Example 1: Test Execution Server

@server.tool()
async def run_test_suite(
    test_path: str = ".",
    pattern: str = "test_*.py",
    verbose: bool = False
) -> Dict[str, Any]:
    """Execute test suite with detailed reporting.
    
    Returns test results, coverage metrics, and failure analysis.
    """
    proc = await asyncio.create_subprocess_exec(
        "pytest", test_path, 
        "-k", pattern,
        "--json-report", 
        "--json-report-file=/tmp/report.json",
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
    )
    
    stdout, stderr = await proc.communicate()
    
    with open("/tmp/report.json") as f:
        report = json.load(f)
    
    return {
        "summary": {
            "total": report["summary"]["total"],
            "passed": report["summary"]["passed"],
            "failed": report["summary"]["failed"],
            "skipped": report["summary"]["skipped"]
        },
        "failures": [
            {
                "test": test["nodeid"],
                "error": test["call"]["longrepr"],
                "traceback": test["call"]["traceback"]
            }
            for test in report["tests"]
            if test["outcome"] == "failed"
        ],
        "execution_time": report["duration"],
        "coverage": await get_coverage_report() if exists(".coverage") else None
    }

Example 2: Database Analytics Server

class DatabaseAnalyticsServer:
    def __init__(self, connection_string: str):
        self.pool = create_async_pool(connection_string, max_connections=10)
    
    @server.tool()
    async def analyze_query_performance(
        self, 
        query: str,
        explain: bool = True
    ) -> Dict[str, Any]:
        """Analyze query performance with execution plan."""
        async with self.pool.acquire() as conn:
            # Get execution plan
            if explain:
                plan = await conn.fetch(f"EXPLAIN ANALYZE {query}")
                
            # Execute actual query with timing
            start = time.perf_counter()
            results = await conn.fetch(query)
            execution_time = time.perf_counter() - start
            
            return {
                "row_count": len(results),
                "execution_time_ms": execution_time * 1000,
                "execution_plan": plan if explain else None,
                "index_usage": self._analyze_index_usage(plan),
                "optimization_suggestions": self._suggest_optimizations(plan, execution_time)
            }
    
    def _analyze_index_usage(self, plan: List[Dict]) -> Dict:
        """Extract index usage statistics from execution plan."""
        # Implementation details...
        pass

Example 3: Code Analysis Server

@server.resource(
    uri="code://analysis/{file_path:path}"
)
async def analyze_code_complexity(file_path: str) -> Resource:
    """Provide code complexity analysis with actionable metrics."""
    
    ast_tree = ast.parse(read_file(file_path))
    
    analysis = {
        "complexity": {
            "cyclomatic": calculate_cyclomatic_complexity(ast_tree),
            "cognitive": calculate_cognitive_complexity(ast_tree),
            "halstead": calculate_halstead_metrics(ast_tree)
        },
        "structure": {
            "classes": extract_classes(ast_tree),
            "functions": extract_functions(ast_tree),
            "dependencies": extract_imports(ast_tree)
        },
        "quality_metrics": {
            "maintainability_index": calculate_maintainability_index(ast_tree),
            "test_coverage": await get_coverage_for_file(file_path),
            "documentation_ratio": calculate_doc_ratio(ast_tree)
        },
        "recommendations": generate_refactoring_suggestions(analysis)
    }
    
    return Resource(
        content=analysis,
        metadata={
            "analyzer_version": "2.1.0",
            "timestamp": datetime.utcnow().isoformat(),
            "file_hash": calculate_file_hash(file_path)
        }
    )

Technical Benefits

Protocol Standardization

  • Language-agnostic implementation through JSON-RPC 2.0
  • Consistent capability discovery across different LLM providers
  • Versioned protocol specification for backward compatibility

Security Architecture

  • Explicit capability grants with fine-grained permissions
  • Sandboxed execution environments for untrusted operations
  • Audit logging for compliance and debugging

Performance Optimization

  • Connection pooling for efficient resource utilization
  • Async/await patterns for concurrent operations
  • Response caching with intelligent invalidation

Contextual Intelligence

  • Project-specific knowledge through resource exposure
  • Historical context preservation across sessions
  • Semantic understanding of domain-specific operations

Team Collaboration

  • Shared prompt libraries for consistent workflows
  • Centralized tool repositories for team access
  • Version-controlled server configurations

Extensibility

  • Plugin architecture for custom capability types
  • Middleware support for cross-cutting concerns
  • WebSocket support for real-time streaming operations

Architecture Summary

MCP establishes a client-server protocol for LLM-system integration:

  • Tools provide executable functions with typed parameters and structured returns
  • Resources expose data sources with caching and access control
  • Prompts define reusable workflows with context requirements

The protocol maintains clear separation between AI capabilities and human oversight, implementing the 80-20 principle where automation handles routine operations while humans retain control over critical decisions.

Key Technical Characteristics

  • Transport-agnostic: Supports stdio, HTTP(S), WebSocket
  • Language-independent: JSON-RPC 2.0 for universal compatibility
  • Async-first: Non-blocking operations for scalability
  • Stateful connections: Session management and context preservation
  • Capability negotiation: Dynamic feature discovery and versioning

Further Reading

MCP: Structured integration between language models and external systems.