Cómo construir un servidor MCP: Paso a paso con ejemplos de código

El Protocolo de Contexto de Modelo (MCP) se está convirtiendo en el nuevo estándar para conectar modelos de IA con herramientas y servicios del mundo real. Construir un servidor MCP te permite exponer datos, acciones y recursos a un LLM como Claude a través de una interfaz simple y estandarizada.

En esta guía, aprenderás paso a paso cómo configurar un servidor MCP básico en Python, definir recursos y herramientas, y conectarlo a un cliente MCP.

Qué es un servidor MCP

Un servidor MCP actúa como puente entre un LLM y un sistema externo como una base de datos, almacenamiento de archivos o API. Define recursos (datos legibles), herramientas (acciones) e indicaciones (instrucciones) de manera que el LLM pueda utilizarlos de forma segura durante sus tareas.

En lugar de escribir una integración personalizada para cada modelo o herramienta, MCP ofrece un estándar universal que funciona con la versión 0.1 del protocolo (actual a abril de 2025).

Lo que necesitas antes de empezar

• Python 3.8 o posterior
• Experiencia básica con scripts de Python
• SDK de MCP para Python (disponible vía pip)
• Un cliente compatible con MCP como Claude Desktop o Cursor (opcional para pruebas)
• Git para control de versiones (recomendado)
• Un editor de texto o IDE (Visual Studio Code recomendado)

Entendiendo la estructura central

En MCP:

• Servidor: Proporciona recursos y herramientas al LLM.
• Cliente: Conecta el LLM a tu servidor.
• Protocolo: Gestiona la comunicación entre cliente y servidor.

Definirás dos primitivas importantes:

• Recurso: Información estática o dinámica que el LLM puede leer.
• Herramienta: Una función ejecutable que el LLM puede invocar.

El flujo de comunicación funciona de la siguiente manera:

1. El LLM (a través de un cliente) solicita datos o acciones de tu servidor
2. Tu servidor procesa estas solicitudes y devuelve respuestas estandarizadas
3. El LLM puede entonces usar esta información en su razonamiento y respuestas

1. Configura tu proyecto Python

Comienza creando un directorio para el proyecto y un entorno virtual de Python.

mkdir my_mcp_server
cd my_mcp_server
python -m venv venv
source venv/bin/activate  # Linux/Mac
venvScriptsactivate     # Windows

Crea una estructura básica de proyecto:

mkdir -p src/resources src/tools tests
touch src/__init__.py src/resources/__init__.py src/tools/__init__.py
touch requirements.txt README.md

Agrega lo siguiente a tu requirements.txt:

mcp-server>=0.1.0
pydantic>=2.0.0
pytest>=7.0.0

2. Instala el SDK de MCP

Instala el SDK del servidor MCP para Python y otras dependencias:

pip install -r requirements.txt

Si el SDK oficial aún no está publicado, es posible que necesites instalarlo desde un repositorio de GitHub:

pip install git+https://github.com/anthropic/mcp-server-python.git

3. Crea un servidor MCP básico

Crea un archivo llamado src/server.py:

from typing import Dict, Any
from mcp_server import MCPServer
import logging

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("mcp_server")

def main() -> None:
    """Initialize and start the MCP server."""
    try:
        server = MCPServer(
            name="MyMCPServer",
            version="0.1.0",
            description="A simple MCP server example"
        )
        
        # Resources and tools will be added here
        
        logger.info("Starting MCP server...")
        server.start()
    except Exception as e:
        logger.error(f"Failed to start MCP server: {e}")
        raise

if __name__ == "__main__":
    main()

Esto configura un servidor MCP básico con registro adecuado.

4. Define un recurso

Los recursos exponen datos que el modelo puede leer. Vamos a crear un archivo src/resources/user_profiles.py:

from typing import List, Dict, Any
from pydantic import BaseModel
import logging

logger = logging.getLogger("mcp_server.resources")

class UserProfile(BaseModel):
    """Data model for user profiles."""
    name: str
    role: str
    department: str = "General"
    years_experience: int = 0

def fetch_user_profiles() -> List[Dict[str, Any]]:
    """
    Fetch user profiles from the database.
    
    Returns:
        List[Dict[str, Any]]: A list of user profile dictionaries.
    """
    try:
        # In a real implementation, this would query a database
        # For this example, we'll return mock data
        users = [
            UserProfile(name="Alice", role="Engineer", department="Engineering", years_experience=5),
            UserProfile(name="Bob", role="Product Manager", department="Product", years_experience=3),
            UserProfile(name="Charlie", role="Designer", department="Design", years_experience=7)
        ]
        
        logger.info(f"Successfully fetched {len(users)} user profiles")
        return [user.model_dump() for user in users]
    except Exception as e:
        logger.error(f"Error fetching user profiles: {e}")
        # In production, you might want to return an empty list or raise
        # a specific exception depending on your error handling strategy
        return []

Ahora actualiza src/server.py para incluir este recurso:

from typing import Dict, Any
from mcp_server import MCPServer, Resource
import logging
from src.resources.user_profiles import fetch_user_profiles

# ... existing code ...

def main() -> None:
    """Initialize and start the MCP server."""
    try:
        server = MCPServer(
            name="MyMCPServer",
            version="0.1.0",
            description="A simple MCP server example"
        )
        
        # Add the user profiles resource
        user_profiles = Resource(
            name="user_profiles",
            description="List of user profiles from the company database.",
            fetch_fn=fetch_user_profiles
        )
        
        server.add_resource(user_profiles)
        
        logger.info("Starting MCP server...")
        server.start()
    except Exception as e:
        logger.error(f"Failed to start MCP server: {e}")
        raise

if __name__ == "__main__":
    main()

El LLM ahora puede consultar user_profiles a través del cliente MCP.

5. Define una herramienta

Las herramientas permiten al LLM ejecutar una acción. Crea un archivo src/tools/user_management.py:

from typing import Dict, Any, Optional
from pydantic import BaseModel, Field, ValidationError
import logging

logger = logging.getLogger("mcp_server.tools")

class CreateUserRequest(BaseModel):
    """Validation model for user creation requests."""
    name: str = Field(..., min_length=2, description="User's full name")
    role: str = Field(..., min_length=2, description="User's job role")
    department: Optional[str] = Field("General", description="User's department")
    years_experience: Optional[int] = Field(0, ge=0, description="Years of professional experience")

def create_user_profile(request_data: Dict[str, Any]) -> Dict[str, Any]:
    """
    Create a new user profile in the database.
    
    Args:
        request_data (Dict[str, Any]): User data containing name, role, etc.
        
    Returns:
        Dict[str, Any]: Response with status and user info
    """
    try:
        # Validate the input data
        user_data = CreateUserRequest(**request_data)
        
        # In a real implementation, this would insert into a database
        # For this example, we'll just log the action
        logger.info(f"Creating new user: {user_data.name} - {user_data.role} in {user_data.department}")
        
        # Return success response with created user data
        return {
            "status": "success",
            "message": f"User {user_data.name} created successfully",
            "user": user_data.model_dump()
        }
    except ValidationError as e:
        # Handle validation errors
        logger.error(f"Validation error: {e}")
        return {
            "status": "error",
            "message": "Invalid user data provided",
            "details": str(e)
        }
    except Exception as e:
        # Handle other errors
        logger.error(f"Error creating user: {e}")
        return {
            "status": "error",
            "message": "Failed to create user",
            "details": str(e)
        }

Ahora actualiza src/server.py para incluir esta herramienta:

from typing import Dict, Any
from mcp_server import MCPServer, Resource, Tool
import logging
from src.resources.user_profiles import fetch_user_profiles
from src.tools.user_management import create_user_profile

# ... existing code ...

def main() -> None:
    """Initialize and start the MCP server."""
    try:
        server = MCPServer(
            name="MyMCPServer",
            version="0.1.0",
            description="A simple MCP server example"
        )
        
        # Add the user profiles resource
        user_profiles = Resource(
            name="user_profiles",
            description="List of user profiles from the company database.",
            fetch_fn=fetch_user_profiles
        )
        
        # Add the create user tool
        create_user = Tool(
            name="create_user_profile",
            description="Create a new user profile in the database.",
            parameters={
                "name": {"type": "string", "description": "User's full name"},
                "role": {"type": "string", "description": "User's job role"},
                "department": {"type": "string", "description": "User's department (optional)"},
                "years_experience": {"type": "integer", "description": "Years of experience (optional)"}
            },
            execute_fn=create_user_profile
        )
        
        server.add_resource(user_profiles)
        server.add_tool(create_user)
        
        logger.info("Starting MCP server...")
        server.start()
    except Exception as e:
        logger.error(f"Failed to start MCP server: {e}")
        raise

6. Manejo de errores y validación

Crea un archivo src/utils/validation.py para centralizar tu lógica de validación:

from typing import Dict, Any, List, Optional, Type
from pydantic import BaseModel, ValidationError
import logging

logger = logging.getLogger("mcp_server.validation")

def validate_request(
    data: Dict[str, Any],
    model_class: Type[BaseModel]
) -> tuple[Optional[BaseModel], Optional[Dict[str, Any]]]:
    """
    Validate request data against a Pydantic model.
    
    Args:
        data: The input data to validate
        model_class: The Pydantic model class to use for validation
        
    Returns:
        tuple: (validated_model, error_dict)
            - If valid: (model instance, None)
            - If invalid: (None, error dictionary)
    """
    try:
        validated_data = model_class(**data)
        return validated_data, None
    except ValidationError as e:
        errors = e.errors()
        error_dict = {
            "status": "error",
            "message": "Validation failed",
            "errors": errors
        }
        logger.error(f"Validation error: {errors}")
        return None, error_dict

Esta función de utilidad puede usarse en todas tus herramientas para validar datos de entrada de manera consistente.

7. Ejecuta y prueba el servidor MCP

Crea un script de prueba simple test_server.py para verificar que tu servidor funciona:

import requests
import json
import time
import subprocess
import sys
from pathlib import Path

def test_server():
    """Simple test to verify the MCP server is running correctly."""
    # Start the server in a separate process
    server_process = subprocess.Popen([sys.executable, "src/server.py"])
    
    try:
        # Wait for server to start
        time.sleep(2)
        
        # Test the server using the MCP client
        # In a real test, you would use the MCP client SDK
        # For this example, we'll simulate a client using HTTP requests
        
        # Assuming the server is running on localhost:8000
        base_url = "http://localhost:8000"
        
        # Test fetching resources
        response = requests.get(f"{base_url}/resources/user_profiles")
        assert response.status_code == 200
        data = response.json()
        print("Resource response:", json.dumps(data, indent=2))
        
        # Test executing a tool
        tool_data = {
            "name": "Test User",
            "role": "Tester",
            "department": "QA"
        }
        response = requests.post(
            f"{base_url}/tools/create_user_profile",
            json=tool_data
        )
        assert response.status_code == 200
        data = response.json()
        print("Tool response:", json.dumps(data, indent=2))
        
        print("All tests passed!")
        
    finally:
        # Clean up: terminate the server process
        server_process.terminate()
        server_process.wait()

if __name__ == "__main__":
    test_server()

Ejecuta tu servidor:

python src/server.py

En una terminal separada, podrías ver una salida como esta cuando el servidor está ejecutándose:

2025-04-28 10:15:23 - mcp_server - INFO - Starting MCP server...
2025-04-28 10:15:23 - mcp_server - INFO - Server listening on 0.0.0.0:8000
2025-04-28 10:15:30 - mcp_server.resources - INFO - Successfully fetched 3 user profiles
2025-04-28 10:15:35 - mcp_server.tools - INFO - Creating new user: Test User - Tester in QA

Luego configura tu cliente MCP (como Claude Desktop) para conectarse a tu servidor MCP local proporcionando la URL del servidor o el comando para iniciar el servidor.

Consideraciones de seguridad

Al implementar un servidor MCP, considera estas mejores prácticas de seguridad:

1. Autenticación: Implementa claves API u OAuth para autenticar clientes.

def authenticate_request(request):
    api_key = request.headers.get("X-API-Key")
    if not api_key or api_key != os.environ.get("MCP_API_KEY"):
        raise ValueError("Invalid API key")

2. Validación de entrada: Siempre valida todas las entradas usando modelos Pydantic.
3. Limitación de tasa: Implementa limitación de tasa para prevenir abusos.
4. HTTPS: Siempre usa HTTPS en producción.
5. Acciones restringidas: Define límites claros sobre lo que las herramientas pueden hacer.

Optimización de rendimiento

1. Caché: Almacena en caché las consultas costosas de recursos:

from functools import lru_cache

@lru_cache(maxsize=128, ttl=300)  # Cache for 5 minutes
def fetch_user_profiles():
    # Expensive database query
    pass

2. Procesamiento asíncrono: Usa async para operaciones limitadas por E/S:

async def fetch_user_profiles():
    async with aiohttp.ClientSession() as session:
        async with session.get("https://api.example.com/users") as response:
            data = await response.json()
            return data

3. Agrupación de conexiones: Usa pools de conexiones para acceso a bases de datos.

Despliegue

Desarrollo local

Para desarrollo local, ejecuta:

python src/server.py

Despliegue con Docker

Crea un Dockerfile:

FROM python:3.10-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["python", "src/server.py"]

Construye y ejecuta:

docker build -t mcp-server .
docker run -p 8000:8000 mcp-server

Despliegue en la nube (AWS)

1. Crea una instancia EC2 o usa AWS App Runner
2. Despliega tu contenedor Docker
3. Configura un Application Load Balancer
4. Configura grupos de seguridad para restringir el acceso

Probando tu servidor MCP

Crea un archivo de prueba tests/test_resources.py:

import pytest
from src.resources.user_profiles import fetch_user_profiles

def test_fetch_user_profiles():
    """Test that user profiles can be fetched successfully."""
    profiles = fetch_user_profiles()
    
    # Check structure
    assert isinstance(profiles, list)
    assert len(profiles) > 0
    
    # Check content
    first_profile = profiles[0]
    assert "name" in first_profile
    assert "role" in first_profile
    assert isinstance(first_profile["name"], str)

Ejecuta las pruebas con:

pytest

Errores comunes y solución de problemas

Problema Solución Ejemplo No se puede conectar al servidor MCP Verifica que tu servidor esté ejecutándose y el puerto sea correcto netstat -tulpn | grep 8000 El LLM no puede encontrar recursos Verifica que los campos name y description estén correctamente configurados Revisa tu inicialización de Resource Errores en la ejecución de herramientas Valida que los parámetros de entrada coincidan con los tipos esperados Usa Pydantic para validación El cliente no puede analizar la salida Asegúrate de que tus funciones devuelvan datos serializables a JSON Usa .model_dump() en lugar de objetos personalizados El servidor se bloquea al iniciar Revisa tus importaciones y variables de entorno Establece DEBUG=True para registro detallado Tiempo de espera de herramienta Agrega manejo de tiempo de espera para llamadas a API externas Usa asyncio.wait_for() con un tiempo límite Fallos de autenticación Verifica claves API y permisos Revisa los encabezados de solicitud Errores de análisis XML/JSON Usa encabezados de tipo de contenido adecuados Establece Content-Type: application/json

Próximos pasos

Después de construir tu servidor MCP básico, considera estas extensiones avanzadas:

1. Integración con bases de datos: Conéctate a PostgreSQL, MongoDB u otras bases de datos.
2. Operaciones de archivos: Agrega herramientas para lectura, escritura y transformación de archivos.
3. APIs externas: Intégrate con servicios populares como GitHub, Slack o Google Drive.
4. Webhooks: Permite que el LLM desencadene eventos en otros sistemas.
5. Recursos de streaming: Admite la transmisión de grandes conjuntos de datos.
6. Acciones conscientes del contexto: Agrega herramientas que entiendan el contexto actual del LLM.

Ejemplo: Agregar una conexión a base de datos

import psycopg2
from contextlib import contextmanager

@contextmanager
def get_db_connection():
    """Create a database connection context manager."""
    conn = None
    try:
        conn = psycopg2.connect(
            host=os.environ.get("DB_HOST"),
            database=os.environ.get("DB_NAME"),
            user=os.environ.get("DB_USER"),
            password=os.environ.get("DB_PASSWORD")
        )
        yield conn
    finally:
        if conn is not None:
            conn.close()

def fetch_user_profiles_from_db():
    """Fetch user profiles from a PostgreSQL database."""
    with get_db_connection() as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT name, role, department FROM users")
            columns = [desc[0] for desc in cur.description]
            return [dict(zip(columns, row)) for row in cur.fetchall()]

Conclusión

Construir un servidor MCP simple en Python abre la puerta para hacer que los LLMs sean mucho más potentes. Al exponer datos y acciones a través de un protocolo limpio y estandarizado, facilitas que los sistemas de IA interactúen de manera segura y significativa con servicios externos.

Comienza con algo pequeño, céntrate en un recurso y una herramienta, y podrás expandirte con el tiempo hacia casos de uso más avanzados como bases de datos, almacenamiento en la nube o APIs internas.

El ecosistema MCP está creciendo rápidamente, y implementar estos estándares ahora posicionará tus aplicaciones para beneficiarse de las mejoras tanto en el protocolo como en los LLMs que lo utilizan.

Preguntas frecuentes