16.5 Building Custom MCP Servers

Course: Claude Code - Enterprise Development

Section: Advanced MCP Integrations

Video Length: 4-5 minutes

Presenter: Daniel Treasure

Opening Hook

"So far, we've connected Claude to existing services via official MCP adapters. But what about your proprietary systems—internal APIs, legacy databases, custom tools? Today, we're building a custom MCP server to safely expose internal systems to Claude without leaking credentials or bypassing security."

Key Talking Points

What to say:

"Not every system has an official MCP adapter—but you can build one."
"Custom MCP servers are the secure bridge between Claude and your internal systems."
"Why build instead of using raw APIs? Security, scope control, credential isolation, audit logging."
"We'll build a simple internal user service MCP and talk about the design principles."
"Components: server process, defined operations, authentication, permission enforcement."

What to show on screen:

Internal API documentation (example)
MCP server code from scratch
Server running and registering with Claude Code
Claude calling MCP operations
Audit logging and permission enforcement

Demo Plan

[00:00 - 01:00] Custom MCP Architecture 1. Explain the three layers: Transport (stdio/HTTP), MCP Protocol (request/response), Your Logic 2. Show: MCP server receives request → validates auth → calls your API → returns sanitized response 3. Draw/show diagram: Claude → MCP Server → Your Internal API → Database 4. Emphasize: Claude never touches credentials, API URLs, or raw data

[01:00 - 02:00] Build Simple MCP Server 1. Create Python file with mcp.server.Server 2. Define first tool: get_user(user_id: str) returns {name, email, status} 3. Show schema definition (input validation) 4. Implement: fetch from internal API with service account credentials 5. Return only necessary fields (strip internal data)

[02:00 - 03:15] Security & Least Privilege 1. Show: MCP server has ONE credential (service account) → Claude never sees it 2. Demonstrate: Input validation (only allow numeric user IDs, prevent SQL injection) 3. Audit logging: every MCP call logged with: timestamp, operation, user_id, result 4. Show error handling: never leak stack traces to Claude, return safe error messages

[03:15 - 04:00] Register & Test MCP Server 1. Start custom MCP server (show process running) 2. Run claude mcp add --transport stdio mycustom -- python server.py 3. In Claude Code, ask: "Get details for user 123" 4. Show Claude calling MCP tool, returning sanitized data 5. Ask: "Get details for user 123; DROP TABLE users" 6. Show MCP validation blocking malicious input

[04:00 - 04:30] Best Practices 1. Least privilege: expose only what Claude needs 2. Input validation: strict type checking, range limits 3. Output sanitization: strip internal fields, no error details 4. Audit everything: log all operations for compliance

Code Examples & Commands

Python MCP Server (complete example):

import os
import json
import logging
from mcp.server import Server
from mcp.types import Tool, TextContent
import httpx
import logging

# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Initialize server
server = Server("internal-user-service")
internal_api_url = os.getenv("INTERNAL_API_URL")
service_account_token = os.getenv("INTERNAL_API_TOKEN")

@server.tool(
    name="get_user",
    description="Get user details by ID",
    inputSchema={
        "type": "object",
        "properties": {
            "user_id": {
                "type": "string",
                "description": "Numeric user ID",
                "pattern": "^[0-9]+$"
            }
        },
        "required": ["user_id"]
    }
)
async def get_user(user_id: str) -> TextContent:
    try:
        # Log the request
        logger.info(f"MCP get_user called with user_id={user_id}")

        # Validate input
        if not user_id.isdigit():
            return TextContent(text="Error: Invalid user ID format")

        # Call internal API with service credentials
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{internal_api_url}/users/{user_id}",
                headers={"Authorization": f"Bearer {service_account_token}"}
            )

        if response.status_code == 404:
            return TextContent(text="User not found")

        response.raise_for_status()
        data = response.json()

        # Sanitize output: return only safe fields
        safe_data = {
            "id": data["id"],
            "name": data["name"],
            "email": data["email"],
            "status": data["status"],
            "department": data.get("department")
        }

        logger.info(f"Successfully retrieved user {user_id}")
        return TextContent(text=json.dumps(safe_data, indent=2))

    except Exception as e:
        logger.error(f"Error in get_user: {str(e)}")
        return TextContent(text="Error: Unable to retrieve user")

@server.tool(
    name="list_users_by_department",
    description="List users in a department",
    inputSchema={
        "type": "object",
        "properties": {
            "department": {
                "type": "string",
                "enum": ["engineering", "product", "sales", "operations"]
            }
        },
        "required": ["department"]
    }
)
async def list_users_by_department(department: str) -> TextContent:
    try:
        logger.info(f"MCP list_users_by_department called with department={department}")

        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{internal_api_url}/departments/{department}/users",
                headers={"Authorization": f"Bearer {service_account_token}"}
            )

        response.raise_for_status()
        data = response.json()

        # Sanitize: return only names and IDs
        safe_data = [
            {"id": u["id"], "name": u["name"]}
            for u in data.get("users", [])
        ]

        return TextContent(text=json.dumps(safe_data, indent=2))

    except Exception as e:
        logger.error(f"Error in list_users_by_department: {str(e)}")
        return TextContent(text="Error: Unable to list users")

if __name__ == "__main__":
    server.run(transport="stdio")

Register & test:

# Set environment variables
export INTERNAL_API_URL="https://api.internal.example.com"
export INTERNAL_API_TOKEN="serviceaccount-xxx"

# Add to Claude Code
claude mcp add --transport stdio user-service -- python /path/to/server.py

# Test in Claude Code:
# "List all engineering team members"
# "Get details for user 5"

Gotchas & Tips

Gotcha 1: Credential Leakage - Never store credentials in code, config files, or share them with Claude - Solution: Use environment variables, secrets managers (.env never in repo) - In demo: show .env file on screen, but don't expand it (keep tokens hidden)

Gotcha 2: Missing Input Validation - If you don't validate input, Claude might send unexpected data - Example: user_id as string instead of int, special characters, very large IDs - Solution: Define strict inputSchema and validate before calling internal API

Gotcha 3: Overly Broad Permissions - Don't expose your entire API—expose only what Claude needs - Bad: "get_anything(path: str)" → Claude can call /admin/secrets - Good: "get_user(user_id: str)" → only users, only by ID

Gotcha 4: Error Message Information Leakage - Don't return stack traces or internal error details to Claude - Bad: "ValueError: Connection refused at api.internal.com:5432" - Good: "Error: Unable to retrieve user"

Tip 1: Schema Drives Behavior - Use inputSchema enums to restrict choices - Example: department = ["eng", "product", "sales"] → Claude can't request invalid departments

Tip 2: Pagination for Large Results - If Claude asks for "all users," return paginated results - Schema: limit (default 50, max 100), offset - Claude will iterate if needed

Tip 3: Versioning & Backward Compatibility - Plan for schema changes early - Example: version your tools (get_user_v2) to avoid breaking Claude workflows

Tip 4: Rate Limiting - Implement backoff in your MCP server if internal API is slow - Prevent Claude from hammering your API

Tip 5: Audit Logging is Gold - Log every MCP call: who (service account), what (operation), when, result - Useful for: debugging, compliance, detecting misuse

Lead-out

"You've now built a production-ready MCP server—the secure bridge between Claude and your internal systems. Security, validation, and logging are built in. We've covered reading external systems (16.1-16.4) and building custom adapters (16.5). Next section: we shift from CLI to SDK—building Claude agents programmatically in your applications."

Reference URLs

Anthropic MCP Specification: https://modelcontextprotocol.io/
MCP Python SDK: https://github.com/anthropics/python-sdk/tree/main/mcp
Example MCP Servers: https://github.com/anthropics/mcp-servers
Security Best Practices for MCPs: [would be in official docs]

Prep Reading

Review MCP specification (basics, 10 min)
Identify internal API you'd like to expose via MCP (5 min)
Read example MCP server code in anthropics/mcp-servers (5 min)

Notes for Daniel

Code pace: This is dense—go slow on the code. Let viewers read each line before moving to the next section.
Highlight security: Emphasize repeatedly that this design prevents Claude from ever seeing credentials or calling APIs directly.
Real example: If possible, use an example related to your own infrastructure (HR system, asset tracking, etc.)
Demo confidence: Practice the MCP server setup before recording. Network issues are common in live demos.
Tone: Position MCP as the "right way" to integrate Claude with internal systems. Not a hack, but a deliberate architecture choice.
Wrap-up: Mention that MCP servers can be deployed as microservices, Docker containers, or serverless functions. This sets up 17.10 (Hosting & Deployment).

Quick Reference