510-mcp-servers.mdc

---
title: MCP Server Engineering Ruleset
description: Patterns and best practices for building Model Context Protocol (MCP) servers and tools.
priority: 510
alwaysApply: false
files:
  include:
    - "**/mcp/**/*.ts"
    - "**/mcp/**/*.py"
    - "**/mcp/**/*.js"
    - "**/*mcp*.ts"
    - "**/*mcp*.py"
---

# MCP Server Engineering Ruleset

**Goal:** Build MCP servers with rich, AI-agent-friendly documentation that enables effective tool selection and usage.

## About MCP

The [Model Context Protocol (MCP)](https://github.blog/open-source/maintainers/mcp-joins-the-linux-foundation-what-this-means-for-developers-building-the-next-era-of-ai-tools-and-agents/) is a **vendor-neutral, open standard** managed by the Linux Foundation's Agentic AI Foundation. MCP enables consistent AI agent integration across platforms (Claude Code, Cursor, GitHub Copilot, and others), solving the n×m integration problem where every AI client had to integrate separately with every tool.

**Key Capabilities:**
- **Multi-vendor:** GitHub, Microsoft, OpenAI, Anthropic, community-supported
- **Enterprise-ready:** OAuth for remote servers, no proprietary tokens
- **Long-running tasks:** Support for builds, deployments, multi-minute operations
- **Discoverable:** MCP Registry for governance and discoverability
- **Production-grade:** Industry-standard maturity (like Kubernetes, GraphQL)

## Core Principle

**Exception to lean comment rules:** MCP tools require **detailed docstrings** because AI agents rely on them for tool selection and usage. Unlike human-facing code, agents need rich descriptions to understand capabilities and constraints. This is critical for vendor-neutral AI agent ecosystems.

## Tool Documentation Standards

### Docstring Structure

MCP tool docstrings serve as API documentation for AI agents. They must be comprehensive:

```python
@mcp.tool
async def kubectl(args: str) -> str:
    """Execute kubectl commands for Kubernetes cluster management.

    Use this tool for all kubectl operations. Provide the kubectl
    arguments as a single string (e.g., "get pods -n default").

    **Important:** This tool requires kubectl to be installed and
    configured with cluster access. Commands are executed with the
    current user's kubeconfig context.

    Args:
        args: kubectl command arguments as a single string.
              Examples:
              - "get pods -n default"
              - "apply -f deployment.yaml"
              - "logs pod-name -n namespace --tail=100"

    Returns:
        JSON string with structure:
        {
            "stdout": "command output",
            "stderr": "error output (empty if success)",
            "statusCode": 0
        }

        Status codes:
        - 0: Success
        - 1: General error
        - 2: Misuse of command

    Raises:
        RuntimeError: If kubectl is not installed or not in PATH

    Example:
        >>> result = await kubectl("get pods -n default")
        >>> data = json.loads(result)
        >>> print(data["stdout"])
        NAME                    READY   STATUS    RESTARTS   AGE
        pod-123                 1/1     Running   0          5m
    """
```

### Poor Example (Don't Do This)

```python
@mcp.tool
async def kubectl(args: str) -> str:
    """Run kubectl.

    Returns:
        Output: <stdout>
    """
```

**Problems:**
- Too vague
- No usage examples
- No error handling info
- No input format specification
- No output structure details

## Input/Output Documentation

### Specify Input Structure Clearly

```python
@mcp.tool
async def create_resource(
    resource_type: str,
    name: str,
    properties: dict,
    namespace: str = "default"
) -> str:
    """Create a Kubernetes resource.

    Args:
        resource_type: Type of resource (e.g., "deployment", "service", "configmap").
                       Must be a valid Kubernetes resource type.
        name: Name for the resource. Must be a valid DNS subdomain name
              (lowercase alphanumeric and hyphens, max 253 chars).
        properties: Resource specification as a dictionary. Structure depends
                    on resource_type:
                    - For deployments: {"replicas": int, "image": str, "ports": [...]}
                    - For services: {"type": str, "ports": [...]}
                    - For configmaps: {"data": dict}
        namespace: Kubernetes namespace (default: "default").
                   Must exist before creating resources.

    Returns:
        JSON string with creation result:
        {
            "success": bool,
            "resource": {
                "type": "deployment",
                "name": "my-app",
                "namespace": "default"
            },
            "message": "Resource created successfully"
        }
    """
```

### Specify Output Structure Clearly

For command execution tools, return structured JSON:

```python
@mcp.tool
async def execute_command(command: str, timeout: int = 30) -> str:
    """Execute a shell command and return structured output.

    Args:
        command: Shell command to execute. Will be executed in a
                 subprocess with shell=True. Be careful with user input.
        timeout: Maximum execution time in seconds (default: 30).
                 Command will be killed if it exceeds this time.

    Returns:
        JSON string with structure:
        {
            "stdout": "standard output from command",
            "stderr": "standard error output (empty if none)",
            "statusCode": 0,
            "duration": 1.23,
            "timedOut": false
        }

        Status codes follow Unix conventions:
        - 0: Success
        - 1-255: Command-specific error codes

        If command times out:
        {
            "stdout": "",
            "stderr": "Command timed out after 30 seconds",
            "statusCode": 124,
            "duration": 30.0,
            "timedOut": true
        }
    """
```

## Error Handling Documentation

### Document All Error Cases

```python
@mcp.tool
async def read_file(file_path: str) -> str:
    """Read contents of a file.

    Args:
        file_path: Path to file. Can be absolute or relative to
                   current working directory. Supports both Unix
                   and Windows paths.

    Returns:
        File contents as string. Empty string if file is empty.

    Raises:
        FileNotFoundError: If file does not exist
        PermissionError: If file cannot be read due to permissions
        IsADirectoryError: If path points to a directory, not a file

    Example:
        >>> content = await read_file("/path/to/file.txt")
        >>> print(content)
        File contents here...
    """
```

## Tool Naming and Organization

### Use Descriptive Names

```python
#  GOOD: Descriptive name
@mcp.tool
async def get_kubernetes_pods(namespace: str = "default") -> str:
    """Get list of pods in a Kubernetes namespace."""
    pass

#  BAD: Vague name
@mcp.tool
async def get_pods(ns: str = "default") -> str:
    """Get pods."""
    pass
```

### Group Related Tools

```python
# Group Kubernetes-related tools
@mcp.tool
async def k8s_get_pods(namespace: str) -> str:
    """Get Kubernetes pods."""
    pass

@mcp.tool
async def k8s_get_services(namespace: str) -> str:
    """Get Kubernetes services."""
    pass

@mcp.tool
async def k8s_create_deployment(spec: dict) -> str:
    """Create Kubernetes deployment."""
    pass
```

## Type Hints and Validation

### Use Type Hints

```python
from typing import Optional, List, Dict, Any

@mcp.tool
async def process_data(
    data: Dict[str, Any],
    options: Optional[Dict[str, Any]] = None,
    filters: Optional[List[str]] = None
) -> Dict[str, Any]:
    """Process data with optional filters and options.

    Args:
        data: Input data dictionary. Required keys:
              - "id": str
              - "values": List[float]
        options: Optional processing options:
                 - "normalize": bool (default: False)
                 - "threshold": float (default: 0.5)
        filters: Optional list of filter names to apply.
                 Valid filters: ["outliers", "missing", "duplicates"]

    Returns:
        Processed data dictionary with same structure as input,
        plus additional fields:
        - "processed_at": ISO timestamp
        - "filters_applied": List[str]
    """
```

## Examples in Documentation

### Include Multiple Examples

```python
@mcp.tool
async def query_database(query: str, params: Optional[Dict] = None) -> str:
    """Execute a database query safely using parameterized queries.

    **Security:** This tool uses parameterized queries to prevent
    SQL injection. Never concatenate user input into queries.

    Args:
        query: SQL query with placeholders (use ? or :name).
               Example: "SELECT * FROM users WHERE id = ?"
        params: Optional parameters for query placeholders.
                Example: {"id": 123} or [123] for positional placeholders.

    Returns:
        JSON array of result rows:
        [
            {"id": 1, "name": "Alice", "email": "alice@acme.com"},
            {"id": 2, "name": "Bob", "email": "bob@acme.com"}
        ]

        Empty array [] if no results.

    Examples:
        # Simple query
        >>> result = await query_database("SELECT * FROM users LIMIT 10")

        # Parameterized query
        >>> result = await query_database(
        ...     "SELECT * FROM users WHERE id = ?",
        ...     params=[123]
        ... )

        # Named parameters
        >>> result = await query_database(
        ...     "SELECT * FROM users WHERE email = :email",
        ...     params={"email": "user@acme.com"}
        ... )
    """
```

## Security Considerations

### Document Security Implications

```python
@mcp.tool
async def execute_shell_command(command: str) -> str:
    """Execute a shell command.

    **SECURITY WARNING:** This tool executes shell commands with
    the privileges of the MCP server process. Use with extreme
    caution. Never execute commands constructed from untrusted user
    input without validation.

    Args:
        command: Shell command to execute. Will be executed via
                 subprocess with shell=True. Commands are logged
                 for security auditing.

    Returns:
        Command output as JSON:
        {
            "stdout": "...",
            "stderr": "...",
            "exitCode": 0
        }
    """
```

---

## OAuth & Remote Servers

### OAuth Flow for Enterprise MCP Servers

MCP supports OAuth for remote servers, enabling secure, compliant integrations at scale:

```typescript
// OAuth-enabled MCP server configuration
import { Server } from '@modelcontextprotocol/sdk/server/index.js';
import { OAuth2Server } from '@modelcontextprotocol/sdk/oauth';

const server = new Server(
  {
    name: 'enterprise-tools',
    version: '1.0.0',
  },
  {
    capabilities: {
      tools: {},
      oauth: {
        authorizationEndpoint: 'https://auth.acme.com/oauth/authorize',
        tokenEndpoint: 'https://auth.acme.com/oauth/token',
        scopes: ['tools:read', 'tools:execute'],
      },
    },
  }
);

// Validate OAuth token before tool execution
async function validateToken(token: string): Promise<boolean> {
  const response = await fetch('https://auth.acme.com/oauth/verify', {
    headers: { Authorization: `Bearer ${token}` },
  });
  return response.ok;
}
```

**Benefits:**
- No proprietary token formats
- Familiar authentication model for enterprises
- Multi-machine orchestration support
- Compliant with existing enterprise auth stacks

---

## Long-Running Tasks

### Pattern for Builds, Deployments, and Multi-Minute Operations

MCP supports long-running task APIs without polling hacks or custom callbacks:

```python
from typing import Optional
import asyncio
import uuid

class TaskStatus:
    """Track long-running task status."""

    def __init__(self, task_id: str):
        self.id = task_id
        self.status = "pending"  # pending, running, completed, failed
        self.progress = 0.0
        self.message = ""
        self.result: Optional[str] = None
        self.error: Optional[str] = None

# In-memory task store (use Redis/DB in production)
tasks: dict[str, TaskStatus] = {}

@mcp.tool
async def start_deployment(
    app_name: str,
    version: str,
    environment: str
) -> str:
    """Start application deployment (long-running task).

    This operation typically takes 5-15 minutes. Use check_task_status()
    to poll for completion.

    Args:
        app_name: Application to deploy
        version: Version/tag to deploy
        environment: Target environment (dev, staging, prod)

    Returns:
        JSON with task tracking info:
        {
            "task_id": "uuid-here",
            "status": "pending",
            "estimated_duration": "10-15 minutes",
            "check_status_tool": "check_task_status"
        }
    """
    task_id = str(uuid.uuid4())
    tasks[task_id] = TaskStatus(task_id)

    # Start async deployment
    asyncio.create_task(run_deployment(task_id, app_name, version, environment))

    return json.dumps({
        "task_id": task_id,
        "status": "pending",
        "estimated_duration": "10-15 minutes",
        "check_status_tool": "check_task_status"
    })

@mcp.tool
async def check_task_status(task_id: str) -> str:
    """Check status of a long-running task.

    Args:
        task_id: Task ID returned from start_deployment() or similar

    Returns:
        JSON with current status:
        {
            "task_id": "uuid",
            "status": "running",
            "progress": 0.45,
            "message": "Deploying to production (45%)",
            "result": null  // or result data if completed
        }
    """
    if task_id not in tasks:
        return json.dumps({"error": "Task not found"})

    task = tasks[task_id]
    return json.dumps({
        "task_id": task.id,
        "status": task.status,
        "progress": task.progress,
        "message": task.message,
        "result": task.result,
        "error": task.error
    })
```

**Best Practices:**
- Return task ID immediately
- Provide separate status-check tool
- Include estimated duration
- Report progress percentage
- Store task state (Redis, DB, not just memory)

---

## MCP Registry & Discoverability

### Publishing to MCP Registry

The [MCP Registry](https://github.com/modelcontextprotocol/registry) provides discoverability and governance:

```json
// mcp-server.json (registry manifest)
{
  "name": "enterprise-tools",
  "version": "1.0.0",
  "description": "Internal enterprise tools for deployment and monitoring",
  "author": "Platform Team",
  "license": "MIT",
  "repository": "https://github.com/acme/mcp-enterprise-tools",
  "tools": [
    {
      "name": "k8s_deploy",
      "description": "Deploy applications to Kubernetes clusters",
      "category": "deployment"
    },
    {
      "name": "check_service_health",
      "description": "Monitor service health metrics",
      "category": "monitoring"
    }
  ],
  "oauth": {
    "required": true,
    "scopes": ["deploy:write", "monitoring:read"]
  },
  "tags": ["kubernetes", "deployment", "monitoring", "enterprise"]
}
```

**Enterprise Governance:**
- Discoverability: Developers find high-quality servers
- Governance: Enterprises control what users adopt
- Versioning: Track and manage MCP server versions
- Compliance: Audit which tools are available

---

## Security Best Practices

### MCP Server Security Checklist

```python
#  GOOD: Secure MCP server implementation
import secrets
from typing import Optional

# 1. Token Validation
async def validate_request(token: Optional[str]) -> bool:
    """Validate OAuth token or API key."""
    if not token:
        return False

    # Verify with auth provider (OAuth)
    return await verify_oauth_token(token)

# 2. Input Sanitization
@mcp.tool
async def execute_command(command: str) -> str:
    """Execute command with input validation."""

    # Allowlist approach
    ALLOWED_COMMANDS = ['kubectl', 'helm', 'docker']
    cmd_parts = command.split()

    if not cmd_parts or cmd_parts[0] not in ALLOWED_COMMANDS:
        raise ValueError(f"Command not allowed: {cmd_parts[0]}")

    # Additional validation
    if any(dangerous in command for dangerous in ['rm -rf', '> /dev/', '|', ';']):
        raise ValueError("Potentially dangerous command detected")

    return await safe_execute(command)

# 3. Rate Limiting
from collections import defaultdict
import time

request_counts = defaultdict(list)

async def check_rate_limit(client_id: str, limit: int = 100, window: int = 60) -> bool:
    """Implement rate limiting per client."""
    now = time.time()

    # Clean old requests
    request_counts[client_id] = [
        ts for ts in request_counts[client_id]
        if now - ts < window
    ]

    if len(request_counts[client_id]) >= limit:
        return False

    request_counts[client_id].append(now)
    return True

# 4. Audit Logging
import logging

audit_logger = logging.getLogger('mcp.audit')

@mcp.tool
async def sensitive_operation(params: dict) -> str:
    """Log all sensitive operations."""
    audit_logger.info(
        "Tool execution",
        extra={
            "tool": "sensitive_operation",
            "client_id": get_client_id(),
            "params": sanitize_params(params),
            "timestamp": time.time()
        }
    )

    return await execute_operation(params)
```

**Security Principles:**
1. **OAuth by default** for remote servers
2. **Input validation** on all tool parameters
3. **Rate limiting** to prevent abuse
4. **Audit logging** for compliance
5. **Least privilege** - only required scopes
6. **No hardcoded secrets** - use environment variables
7. **Allowlist approach** for commands/operations

---

## Testing MCP Tools

### Test Tool Documentation

```python
def test_tool_docstring_completeness():
    """Verify tool docstrings contain required information."""
    tool = kubectl

    doc = tool.__doc__
    assert "Args:" in doc, "Missing Args section"
    assert "Returns:" in doc, "Missing Returns section"
    assert "Example" in doc or "Examples" in doc, "Missing examples"
    assert "kubectl" in doc.lower(), "Tool name not in docstring"
```

## Review Checklist

When reviewing MCP tool code, check:

### Documentation
- [ ] Docstring is comprehensive and detailed
- [ ] Input parameters are clearly documented
- [ ] Output structure is specified (especially for JSON)
- [ ] Error cases are documented
- [ ] Examples are provided
- [ ] Type hints are used
- [ ] Tool name is descriptive
- [ ] Related tools are grouped logically

### Security
- [ ] Security implications are noted
- [ ] OAuth configuration (if remote server)
- [ ] Input validation and sanitization
- [ ] Rate limiting implemented
- [ ] Audit logging for sensitive operations
- [ ] No hardcoded secrets

### Production Readiness
- [ ] Return format is consistent (structured JSON for commands)
- [ ] Long-running tasks use task ID pattern
- [ ] Error handling with proper status codes
- [ ] Timeout configuration
- [ ] Registry manifest (mcp-server.json) if publishing

---

## Resources

- [MCP Specification](https://spec.modelcontextprotocol.io/)
- [MCP Registry](https://github.com/modelcontextprotocol/registry)
- [MCP joins Linux Foundation](https://github.blog/open-source/maintainers/mcp-joins-the-linux-foundation-what-this-means-for-developers-building-the-next-era-of-ai-tools-and-agents/)
- [MCP SDK (TypeScript/Python)](https://github.com/modelcontextprotocol/sdk)

### Reference Implementations

- **[Shellwright](https://github.com/dwmkerr/shellwright)** (`@dwmkerr/shellwright`) - Terminal automation MCP server demonstrating PTY session management, terminal rendering (SVG/PNG/GIF), HTTP file serving, and comprehensive tool documentation patterns. Excellent example of building MCP servers that interact with system processes.

---

**Purpose**: MCP server patterns, OAuth, long-running tasks, security, and registry publishing