Capítulo 5: Model Context Protocol (MCP)

¿Qué es MCP?

Model Context Protocol (MCP) es un estándar abierto creado por Anthropic que define cómo los modelos de lenguaje se comunican con herramientas y sistemas externos. Es, en esencia, el “USB” de las integraciones de IA: un protocolo universal que permite enchufar cualquier herramienta a cualquier agente.

El problema que resuelve

Antes de MCP, cada proveedor de IA tenía su propio formato para definir herramientas. OpenAI usaba un formato, Anthropic otro, Cohere otro diferente. Los desarrolladores tenían que reimplementar sus integraciones para cada proveedor. MCP estandariza este proceso.

flowchart TB
    subgraph SIN_MCP["Sin MCP (antes)"]
        direction LR
        A1["Claude API"] -->|"formato Anthropic"| T1["Tu herramienta"]
        A2["OpenAI API"] -->|"formato OpenAI"| T1
        A3["Gemini API"] -->|"formato Google"| T1
    end

    subgraph CON_MCP["Con MCP (ahora)"]
        direction LR
        B1["Claude Agent"] -->|"protocolo MCP"| ML["MCP Layer"]
        B2["Cursor IDE"] -->|"protocolo MCP"| ML
        B3["Any LLM"] -->|"protocolo MCP"| ML
        ML --> S1["Servidor MCP 1"]
        ML --> S2["Servidor MCP 2"]
        ML --> S3["Servidor MCP 3"]
    end

Arquitectura cliente-servidor del protocolo

MCP funciona bajo un modelo cliente-servidor donde:

• Host: La aplicación que usa el LLM (Claude Code, tu agente personalizado)
• Client: El conector MCP dentro del host que gestiona las conexiones
• Server: El proceso externo que expone herramientas y recursos

sequenceDiagram
    participant A as Agente (Host)
    participant C as MCP Client
    participant S as MCP Server
    participant E as Sistema Externo

    A->>C: Inicializar sesión
    C->>S: initialize (protocol version, capabilities)
    S-->>C: ServerInfo + capabilities
    C-->>A: Conexión establecida

    A->>C: ¿Qué herramientas hay?
    C->>S: tools/list
    S-->>C: [{name, description, inputSchema}]
    C-->>A: Lista de herramientas disponibles

    A->>C: Ejecutar herramienta "buscar_usuario"
    C->>S: tools/call {name: "buscar_usuario", arguments: {...}}
    S->>E: Consulta al sistema externo
    E-->>S: Resultado
    S-->>C: [{type: "text", text: "..."}]
    C-->>A: Resultado de la herramienta

Diagrama completo: agente con múltiples servidores MCP

flowchart TB
    subgraph AGENT["Proceso del Agente"]
        Claude["Claude Model"]
        MCPClient["MCP Client Layer"]
        Claude <-->|"tool calls / results"| MCPClient
    end

    subgraph STDIO["Servidores stdio (subprocesos)"]
        PW["playwright-mcp\n(automatización web)"]
        GH["github-mcp\n(repositorios)"]
        FS["filesystem-mcp\n(archivos)"]
    end

    subgraph HTTP["Servidores HTTP/SSE (remotos)"]
        SLACK["slack-mcp\n:3001"]
        GDRIVE["gdrive-mcp\n:3002"]
        CUSTOM["custom-api-mcp\n:8080"]
    end

    subgraph INPROC["Servidor In-Process (tu código)"]
        DB["db_tools\n(SQLite/PostgreSQL)"]
        CACHE["cache_tools\n(Redis)"]
        BIZ["business_logic_tools"]
    end

    MCPClient -->|"stdin/stdout"| PW
    MCPClient -->|"stdin/stdout"| GH
    MCPClient -->|"stdin/stdout"| FS
    MCPClient -->|"HTTP SSE"| SLACK
    MCPClient -->|"HTTP SSE"| GDRIVE
    MCPClient -->|"HTTP SSE"| CUSTOM
    MCPClient -->|"llamada directa"| DB
    MCPClient -->|"llamada directa"| CACHE
    MCPClient -->|"llamada directa"| BIZ

Transporte: stdio vs HTTP/SSE

MCP define dos mecanismos de transporte principales:

stdio (Standard Input/Output):

• El servidor MCP corre como subproceso del host
• Comunicación vía stdin/stdout usando JSON-RPC 2.0
• Ideal para herramientas locales (filesystem, shell, bases de datos locales)
• Latencia muy baja (proceso local)
• El ciclo de vida del servidor está ligado al del host

HTTP/SSE (Server-Sent Events):

• El servidor corre como proceso independiente o servicio remoto
• Comunicación vía HTTP: GET para SSE (mensajes del server), POST para peticiones del client
• Ideal para servicios compartidos, integraciones cloud, servidores persistentes
• Permite múltiples clientes conectados simultáneamente
• Requiere gestión explícita de autenticación

flowchart LR
    subgraph STDIO_FLOW["Transporte stdio"]
        H1["Host"] -->|"JSON-RPC via stdin"| S1["Servidor MCP\n(subproceso)"]
        S1 -->|"respuesta via stdout"| H1
    end

    subgraph HTTP_FLOW["Transporte HTTP/SSE"]
        H2["Host"] -->|"POST /messages"| S2["Servidor MCP\n(servicio HTTP)"]
        S2 -->|"GET /sse (eventos)"| H2
    end

Descubrimiento de herramientas (tool listing)

Cuando el cliente se conecta a un servidor MCP, ejecuta una negociación de capacidades:

1. initialize: intercambia versiones de protocolo y capacidades soportadas
2. tools/list: obtiene la lista completa de herramientas disponibles con sus esquemas
3. resources/list (opcional): lista recursos accesibles (archivos, URLs, etc.)
4. prompts/list (opcional): plantillas de prompts reutilizables

Cada herramienta se describe con:

• name: identificador único (snake_case)
• description: descripción legible para el LLM (crucial para que sepa cuándo usarla)
• inputSchema: esquema JSON Schema de los parámetros

Diferencia MCP vs tool calling directo

Aspecto	Tool Calling Directo	MCP
Definición	En el código del agente	En servidor separado
Reutilización	Solo en ese agente	Cualquier host compatible
Lenguaje	Mismo que el agente	Cualquier lenguaje
Distribución	Manual	npm, PyPI, etc.
Descubrimiento	Hardcoded	Dinámico via protocol
Versioning	Manual	Protocolo versionado
Testing	Integrado	Aislado e independiente

---

Servidores MCP oficiales

Anthropic y la comunidad han publicado decenas de servidores MCP listos para usar. Aquí los más importantes con configuración completa.

Playwright: automatización de navegador

Permite que tu agente controle un navegador web: navegar URLs, hacer clic, llenar formularios, tomar screenshots, extraer datos.

Instalación:

npm install -g @playwright/mcp
# O usando npx sin instalación global
npx @playwright/mcp --help

Configuración en Python:

from claude_code_sdk import query, ClaudeCodeOptions, MCPServerStdio

options = ClaudeCodeOptions(
    mcp_servers=[
        MCPServerStdio(
            name="playwright",
            command="npx",
            args=["@playwright/mcp", "--browser", "chromium"],
            env={
                "PLAYWRIGHT_HEADLESS": "true"
            }
        )
    ],
    allowed_tools=["mcp__playwright__navigate", "mcp__playwright__screenshot",
                   "mcp__playwright__click", "mcp__playwright__fill"]
)

async for event in query(
    prompt="Ve a https://example.com y dime el título de la página",
    options=options
):
    if hasattr(event, 'type') and event.type == 'result':
        print(event.result)

Configuración en TypeScript:

import { query, ClaudeCodeOptions } from "@anthropic-ai/claude-code-sdk";

const options: ClaudeCodeOptions = {
  mcpServers: [
    {
      name: "playwright",
      command: "npx",
      args: ["@playwright/mcp", "--browser", "chromium"],
      env: { PLAYWRIGHT_HEADLESS: "true" }
    }
  ],
  allowedTools: ["mcp__playwright__navigate", "mcp__playwright__screenshot"]
};

for await (const event of query({
  prompt: "Ve a https://example.com y captura un screenshot",
  options
})) {
  if (event.type === "result") console.log(event.result);
}

GitHub: repositorios, issues, PRs

Acceso completo a la API de GitHub: clonar repos, crear issues, abrir PRs, revisar código, gestionar workflows.

Instalación:

npm install -g @modelcontextprotocol/server-github

Configuración en Python:

from claude_code_sdk import query, ClaudeCodeOptions, MCPServerStdio

options = ClaudeCodeOptions(
    mcp_servers=[
        MCPServerStdio(
            name="github",
            command="npx",
            args=["-y", "@modelcontextprotocol/server-github"],
            env={
                "GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_your_token_here"
            }
        )
    ]
)

async for event in query(
    prompt="Lista los últimos 5 issues abiertos en mi repositorio 'mi-proyecto'",
    options=options
):
    pass

PostgreSQL: base de datos relacional

Lee y escribe en PostgreSQL. Ideal para agentes que necesitan consultar o actualizar datos.

Instalación:

npm install -g @modelcontextprotocol/server-postgres

Configuración:

from claude_code_sdk import query, ClaudeCodeOptions, MCPServerStdio

options = ClaudeCodeOptions(
    mcp_servers=[
        MCPServerStdio(
            name="postgres",
            command="npx",
            args=[
                "-y", "@modelcontextprotocol/server-postgres",
                "postgresql://usuario:password@localhost:5432/mi_db"
            ]
        )
    ]
)

En TypeScript:

const options: ClaudeCodeOptions = {
  mcpServers: [{
    name: "postgres",
    command: "npx",
    args: [
      "-y", "@modelcontextprotocol/server-postgres",
      process.env.DATABASE_URL!
    ]
  }]
};

Filesystem: acceso extendido a archivos

El SDK ya tiene herramientas de archivo built-in, pero el servidor MCP de filesystem ofrece capacidades adicionales como watch de directorios.

options = ClaudeCodeOptions(
    mcp_servers=[
        MCPServerStdio(
            name="filesystem",
            command="npx",
            args=[
                "-y", "@modelcontextprotocol/server-filesystem",
                "/ruta/permitida/1",
                "/ruta/permitida/2"
            ]
        )
    ]
)

Google Drive: documentos y hojas de cálculo

options = ClaudeCodeOptions(
    mcp_servers=[
        MCPServerStdio(
            name="gdrive",
            command="npx",
            args=["-y", "@modelcontextprotocol/server-gdrive"],
            env={
                "GDRIVE_CREDENTIALS_PATH": "/home/user/.config/gdrive-mcp/credentials.json"
            }
        )
    ]
)

Slack: mensajería y notificaciones

options = ClaudeCodeOptions(
    mcp_servers=[
        MCPServerStdio(
            name="slack",
            command="npx",
            args=["-y", "@modelcontextprotocol/server-slack"],
            env={
                "SLACK_BOT_TOKEN": "xoxb-your-bot-token",
                "SLACK_TEAM_ID": "T01234ABCDE"
            }
        )
    ]
)

SQLite: base de datos local

Perfecto para prototipado y datos locales sin necesidad de servidor:

options = ClaudeCodeOptions(
    mcp_servers=[
        MCPServerStdio(
            name="sqlite",
            command="npx",
            args=[
                "-y", "@modelcontextprotocol/server-sqlite",
                "--db-path", "/home/user/data/mi_base.db"
            ]
        )
    ]
)

Brave Search: búsqueda web

options = ClaudeCodeOptions(
    mcp_servers=[
        MCPServerStdio(
            name="brave-search",
            command="npx",
            args=["-y", "@modelcontextprotocol/server-brave-search"],
            env={
                "BRAVE_API_KEY": "BSA_your_api_key_here"
            }
        )
    ]
)

---

Conectar servidor MCP externo (stdio)

Configuración stdio completa

La configuración stdio requiere especificar el comando que lanza el servidor y sus argumentos. El SDK maneja automáticamente el ciclo de vida del proceso.

Python completo con múltiples servidores:

import asyncio
import os
from claude_code_sdk import query, ClaudeCodeOptions, MCPServerStdio

async def main():
    options = ClaudeCodeOptions(
        mcp_servers=[
            # Servidor 1: base de datos
            MCPServerStdio(
                name="postgres",
                command="npx",
                args=[
                    "-y",
                    "@modelcontextprotocol/server-postgres",
                    os.environ["DATABASE_URL"]
                ],
                env={
                    "NODE_ENV": "production",
                    "PGSSL": "true"
                }
            ),
            # Servidor 2: GitHub
            MCPServerStdio(
                name="github",
                command="npx",
                args=["-y", "@modelcontextprotocol/server-github"],
                env={
                    "GITHUB_PERSONAL_ACCESS_TOKEN": os.environ["GITHUB_TOKEN"]
                }
            ),
            # Servidor 3: búsqueda web
            MCPServerStdio(
                name="search",
                command="npx",
                args=["-y", "@modelcontextprotocol/server-brave-search"],
                env={
                    "BRAVE_API_KEY": os.environ["BRAVE_API_KEY"]
                }
            )
        ],
        max_turns=20,
        system_prompt="""Eres un asistente con acceso a:
        - Base de datos PostgreSQL de la empresa
        - API de GitHub para el repositorio principal
        - Búsqueda web via Brave Search

        Usa estas herramientas para responder preguntas técnicas con datos actualizados."""
    )

    resultados = []
    async for event in query(
        prompt="¿Cuántos usuarios se registraron esta semana? ¿Hay issues abiertos relacionados con registro?",
        options=options
    ):
        if hasattr(event, 'type'):
            if event.type == 'assistant':
                for block in event.message.content:
                    if hasattr(block, 'text'):
                        print(f"Agente: {block.text}")
            elif event.type == 'result':
                resultados.append(event.result)

    return resultados

asyncio.run(main())

TypeScript equivalente:

import { query, ClaudeCodeOptions } from "@anthropic-ai/claude-code-sdk";

async function main() {
  const options: ClaudeCodeOptions = {
    mcpServers: [
      {
        name: "postgres",
        command: "npx",
        args: ["-y", "@modelcontextprotocol/server-postgres", process.env.DATABASE_URL!],
        env: { NODE_ENV: "production" }
      },
      {
        name: "github",
        command: "npx",
        args: ["-y", "@modelcontextprotocol/server-github"],
        env: { GITHUB_PERSONAL_ACCESS_TOKEN: process.env.GITHUB_TOKEN! }
      },
      {
        name: "search",
        command: "npx",
        args: ["-y", "@modelcontextprotocol/server-brave-search"],
        env: { BRAVE_API_KEY: process.env.BRAVE_API_KEY! }
      }
    ],
    maxTurns: 20,
    systemPrompt: "Eres un asistente con acceso a base de datos, GitHub y búsqueda web."
  };

  for await (const event of query({
    prompt: "¿Cuántos usuarios se registraron esta semana?",
    options
  })) {
    if (event.type === "assistant") {
      for (const block of event.message.content) {
        if ("text" in block) process.stdout.write(block.text);
      }
    }
  }
}

main().catch(console.error);

Variables de entorno del servidor

Los servidores MCP frecuentemente necesitan credenciales. La forma correcta es pasarlas via el campo env:

MCPServerStdio(
    name="mi-servidor",
    command="node",
    args=["servidor-mcp.js"],
    env={
        # Las variables se pasan SOLO al subproceso del servidor
        # No contaminan el entorno del agente
        "API_KEY": os.environ["MI_API_KEY"],
        "DB_PASSWORD": os.environ["DB_PASSWORD"],
        "LOG_LEVEL": "info",
        # Heredar PATH del sistema para que node/npx funcionen
        "PATH": os.environ["PATH"]
    }
)

IMPORTANTE: Si no especificas env, el subproceso hereda el entorno completo del proceso padre. Si especificas env, solo esas variables estarán disponibles. Siempre incluye PATH si el servidor necesita ejecutar otros programas.

Depurar problemas de conexión

Cuando un servidor MCP falla al conectar, los síntomas son errores como MCP server failed to start o timeouts. Para depurar:

import logging
import asyncio
from claude_code_sdk import query, ClaudeCodeOptions, MCPServerStdio

# Activar logging detallado
logging.basicConfig(level=logging.DEBUG)

async def debug_mcp():
    options = ClaudeCodeOptions(
        mcp_servers=[
            MCPServerStdio(
                name="mi-servidor-problematico",
                command="node",
                args=["servidor.js", "--verbose"],
                env={
                    "PATH": os.environ["PATH"],
                    "DEBUG": "mcp:*"  # Activa debug interno del servidor
                }
            )
        ]
    )

    try:
        async for event in query(
            prompt="prueba de conexión: lista las herramientas disponibles",
            options=options
        ):
            print(f"Evento: {type(event).__name__}")
    except Exception as e:
        print(f"Error de conexión MCP: {e}")
        # Verificar manualmente si el servidor arranca:
        # node servidor.js --verbose
        # Si arranca, el problema es el protocolo
        # Si no arranca, revisar el PATH y las dependencias

asyncio.run(debug_mcp())

Checklist de diagnóstico:

1. ¿El comando command existe en el PATH?
2. ¿Las variables de entorno necesarias están en env?
3. ¿El servidor imprime algo en stderr al arrancar? (indica error de inicialización)
4. ¿El servidor responde al mensaje initialize del protocolo MCP?
5. ¿Las versiones de Node.js/Python son compatibles con el servidor?

---

Conectar servidor MCP HTTP/SSE

Cuándo preferir HTTP vs stdio

flowchart TD
    Q1{"¿El servidor necesita\npersistir entre sesiones?"}
    Q1 -->|Sí| HTTP
    Q1 -->|No| Q2

    Q2{"¿Múltiples agentes\nusan el mismo servidor?"}
    Q2 -->|Sí| HTTP
    Q2 -->|No| Q3

    Q3{"¿El servidor está\nen otro host/cloud?"}
    Q3 -->|Sí| HTTP
    Q3 -->|No| STDIO

    HTTP["Usar HTTP/SSE"]
    STDIO["Usar stdio"]

    style HTTP fill:#4CAF50,color:#fff
    style STDIO fill:#2196F3,color:#fff

Configuración URL

Python:

from claude_code_sdk import query, ClaudeCodeOptions, MCPServerHTTP

options = ClaudeCodeOptions(
    mcp_servers=[
        MCPServerHTTP(
            name="mi-servidor-remoto",
            url="https://mcp.miempresa.com/api",
            headers={
                "Authorization": f"Bearer {os.environ['MCP_API_TOKEN']}",
                "X-Client-ID": "mi-agente-v2"
            }
        )
    ]
)

TypeScript:

import { query, ClaudeCodeOptions } from "@anthropic-ai/claude-code-sdk";

const options: ClaudeCodeOptions = {
  mcpServers: [
    {
      name: "mi-servidor-remoto",
      type: "sse",
      url: "https://mcp.miempresa.com/api",
      headers: {
        Authorization: `Bearer ${process.env.MCP_API_TOKEN}`,
        "X-Client-ID": "mi-agente-v2"
      }
    }
  ]
};

Autenticación con Bearer tokens

import os
from claude_code_sdk import query, ClaudeCodeOptions, MCPServerHTTP

# Patrón recomendado: obtener token fresco antes de cada sesión
async def obtener_token_fresco() -> str:
    # Tu lógica de autenticación
    import httpx
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://auth.miempresa.com/token",
            data={
                "grant_type": "client_credentials",
                "client_id": os.environ["CLIENT_ID"],
                "client_secret": os.environ["CLIENT_SECRET"]
            }
        )
        return response.json()["access_token"]

async def agente_con_auth():
    token = await obtener_token_fresco()

    options = ClaudeCodeOptions(
        mcp_servers=[
            MCPServerHTTP(
                name="empresa-api",
                url="https://mcp.miempresa.com",
                headers={
                    "Authorization": f"Bearer {token}",
                    "Content-Type": "application/json"
                }
            )
        ]
    )

    async for event in query(
        prompt="consulta los datos de ventas del mes actual",
        options=options
    ):
        pass

Cuándo preferir HTTP sobre stdio

• Servidor compartido: múltiples instancias del agente usan el mismo servidor
• Estado persistente: el servidor mantiene sesiones, caché o conexiones entre llamadas
• Acceso remoto: el servidor corre en AWS, GCP, un VPS, etc.
• Team/empresa: un equipo mantiene el servidor centralizado
• Actualizaciones independientes: puedes actualizar el servidor sin tocar el agente

---

create_sdk_mcp_server — Servidor in-process

La opción más poderosa del SDK: crear un servidor MCP que corre dentro del mismo proceso que tu agente. Sin latencia de red, sin serialización extra, con acceso directo a tus objetos Python/TypeScript.

Por qué in-process es mejor para tu código

flowchart LR
    subgraph EXTERNO["Servidor Externo (stdio/HTTP)"]
        AG1["Agente"] -->|"serialize JSON\n+ IPC/network"| SRV["Servidor MCP\n(proceso separado)"]
        SRV -->|"deserialize\n+ execute"| DB1["DB Connection"]
        SRV -->|"serialize result"| AG1
    end

    subgraph INPROC["Servidor In-Process"]
        AG2["Agente"] -->|"llamada directa\n(Python/TS)"| FUNC["Tu función"]
        FUNC -->|"acceso directo"| DB2["DB Connection\n(mismo proceso)"]
        FUNC -->|"retorno directo"| AG2
    end

Ventajas in-process:

• Latencia cero de transporte (llamada de función directa)
• Acceso a objetos ya instanciados (conexiones DB, caches, configuración)
• Mismo contexto de errores (excepciones Python/TS nativas)
• Fácil testing (puedes mockear las funciones directamente)
• No requiere serialización adicional de objetos complejos

Decorator @tool en Python: parámetros y tipos

from claude_code_sdk import query, ClaudeCodeOptions
from claude_code_sdk.mcp import create_sdk_mcp_server, tool
from typing import Optional, List
import asyncio

# Crear el servidor in-process
server = create_sdk_mcp_server("mis-herramientas")

# Herramienta básica: solo texto de retorno
@server.tool(description="Suma dos números enteros")
async def sumar(a: int, b: int) -> str:
    return f"La suma de {a} + {b} = {a + b}"

# Herramienta con parámetros opcionales
@server.tool(description="Busca usuarios por nombre o email")
async def buscar_usuario(
    nombre: Optional[str] = None,
    email: Optional[str] = None,
    limite: int = 10
) -> str:
    if not nombre and not email:
        return "Error: debes proporcionar nombre o email"

    # Tu lógica de búsqueda aquí
    resultados = []  # Simulado
    return f"Encontrados {len(resultados)} usuarios"

# Herramienta con lista de parámetros
@server.tool(description="Crea múltiples registros en lote")
async def crear_registros_lote(
    nombres: List[str],
    categoria: str,
    activo: bool = True
) -> str:
    creados = len(nombres)
    return f"Creados {creados} registros en categoría '{categoria}'"

Esquemas JSON Schema para parámetros (todos los tipos)

El SDK genera automáticamente el JSON Schema a partir de los type hints de Python. Aquí todos los tipos soportados:

from claude_code_sdk.mcp import create_sdk_mcp_server, tool
from typing import Optional, List, Dict, Literal, Union
from enum import Enum

server = create_sdk_mcp_server("tipos-demo")

class EstadoPedido(str, Enum):
    PENDIENTE = "pendiente"
    PROCESANDO = "procesando"
    COMPLETADO = "completado"
    CANCELADO = "cancelado"

@server.tool(description="Demuestra todos los tipos de parámetros")
async def demo_tipos(
    # Tipos básicos
    texto: str,
    numero_entero: int,
    numero_decimal: float,
    booleano: bool,

    # Tipos opcionales (pueden ser None)
    texto_opcional: Optional[str] = None,
    numero_opcional: Optional[int] = None,

    # Listas
    lista_textos: List[str] = [],
    lista_numeros: List[int] = [],

    # Diccionario (objeto JSON)
    metadatos: Dict[str, str] = {},

    # Valores literales (enum de strings)
    modo: Literal["rapido", "preciso", "balanceado"] = "balanceado",

    # Enum
    estado: EstadoPedido = EstadoPedido.PENDIENTE
) -> str:
    return f"Recibido: texto={texto}, modo={modo}, estado={estado.value}"

TypeScript equivalente con tipos explícitos:

import { query, ClaudeCodeOptions, createSdkMcpServer } from "@anthropic-ai/claude-code-sdk";

const server = createSdkMcpServer("mis-herramientas");

server.tool(
  "buscar_usuario",
  "Busca usuarios en la base de datos por nombre o email",
  {
    // JSON Schema explícito en TypeScript
    type: "object" as const,
    properties: {
      nombre: { type: "string", description: "Nombre del usuario a buscar" },
      email: { type: "string", format: "email", description: "Email del usuario" },
      limite: { type: "integer", minimum: 1, maximum: 100, default: 10 }
    },
    required: [] // Ninguno requerido (todos opcionales)
  },
  async (args: { nombre?: string; email?: string; limite?: number }) => {
    const limite = args.limite ?? 10;
    // Tu lógica aquí
    return {
      content: [{ type: "text" as const, text: `Buscando con limite ${limite}` }]
    };
  }
);

Retorno correcto: content array

Todas las herramientas MCP deben retornar un content array. En Python puedes retornar:

from claude_code_sdk.mcp import create_sdk_mcp_server

server = create_sdk_mcp_server("retornos-demo")

# Forma simple: string (el SDK lo convierte automáticamente)
@server.tool(description="Retorno simple como string")
async def herramienta_simple(texto: str) -> str:
    return f"Procesado: {texto}"

# Forma explícita: content array completo
@server.tool(description="Retorno con content array explícito")
async def herramienta_explicita(texto: str) -> dict:
    return {
        "content": [
            {
                "type": "text",
                "text": f"Resultado principal: {texto}"
            },
            {
                "type": "text",
                "text": "Información adicional sobre el resultado"
            }
        ]
    }

# Retorno de error (isError = True)
@server.tool(description="Herramienta que puede fallar")
async def herramienta_con_error(id_usuario: int) -> dict:
    if id_usuario <= 0:
        return {
            "content": [
                {"type": "text", "text": f"Error: ID inválido ({id_usuario}). Debe ser positivo."}
            ],
            "isError": True
        }
    return {"content": [{"type": "text", "text": f"Usuario {id_usuario} encontrado"}]}

Tipos de content: text, image, resource

import base64
from pathlib import Path

server = create_sdk_mcp_server("content-types")

# Content type: text (el más común)
@server.tool(description="Genera texto")
async def generar_texto(prompt: str) -> dict:
    return {
        "content": [{"type": "text", "text": f"Respuesta a: {prompt}"}]
    }

# Content type: image (para screenshots, gráficos)
@server.tool(description="Captura screenshot y lo retorna")
async def capturar_screenshot(url: str) -> dict:
    # Simulamos captura de screenshot
    # En la práctica usarías playwright u otro tool
    imagen_bytes = b"PNG_DATA_AQUI"  # Datos reales del PNG
    imagen_b64 = base64.b64encode(imagen_bytes).decode()

    return {
        "content": [
            {
                "type": "text",
                "text": f"Screenshot de {url} capturado exitosamente"
            },
            {
                "type": "image",
                "data": imagen_b64,
                "mimeType": "image/png"
            }
        ]
    }

# Content type: resource (referencia a archivo o URL)
@server.tool(description="Retorna referencia a un archivo generado")
async def generar_reporte(tipo: str) -> dict:
    ruta_reporte = f"/tmp/reporte_{tipo}.pdf"
    # Genera el archivo...

    return {
        "content": [
            {"type": "text", "text": f"Reporte '{tipo}' generado"},
            {
                "type": "resource",
                "resource": {
                    "uri": f"file://{ruta_reporte}",
                    "mimeType": "application/pdf",
                    "text": f"Reporte de {tipo}"
                }
            }
        ]
    }

Servidor con estado: conexiones a base de datos y caché

import asyncio
import sqlite3
from functools import lru_cache
from claude_code_sdk import query, ClaudeCodeOptions
from claude_code_sdk.mcp import create_sdk_mcp_server
from typing import Optional
import json

class ServidorConEstado:
    """Servidor MCP in-process con estado compartido entre herramientas."""

    def __init__(self, db_path: str):
        self.db_path = db_path
        self._conn: Optional[sqlite3.Connection] = None
        self._cache: dict = {}
        self.server = create_sdk_mcp_server("servidor-con-estado")
        self._registrar_herramientas()

    def _obtener_conexion(self) -> sqlite3.Connection:
        """Conexión lazy: se crea solo cuando se necesita."""
        if self._conn is None:
            self._conn = sqlite3.connect(self.db_path)
            self._conn.row_factory = sqlite3.Row
        return self._conn

    def _registrar_herramientas(self):
        @self.server.tool(description="Busca un producto por ID (con cache)")
        async def buscar_producto(id_producto: int) -> str:
            # Cache hit
            cache_key = f"producto:{id_producto}"
            if cache_key in self._cache:
                return f"[CACHE] {json.dumps(dict(self._cache[cache_key]))}"

            # Cache miss: consultar DB
            conn = self._obtener_conexion()
            cursor = conn.execute(
                "SELECT * FROM productos WHERE id = ?", (id_producto,)
            )
            row = cursor.fetchone()

            if not row:
                return f"Producto {id_producto} no encontrado"

            resultado = dict(row)
            self._cache[cache_key] = resultado  # Guardar en cache
            return json.dumps(resultado)

        @self.server.tool(description="Actualiza el precio de un producto")
        async def actualizar_precio(id_producto: int, nuevo_precio: float) -> str:
            if nuevo_precio < 0:
                return f"Error: el precio no puede ser negativo"

            conn = self._obtener_conexion()
            conn.execute(
                "UPDATE productos SET precio = ? WHERE id = ?",
                (nuevo_precio, id_producto)
            )
            conn.commit()

            # Invalidar cache
            cache_key = f"producto:{id_producto}"
            self._cache.pop(cache_key, None)

            return f"Precio del producto {id_producto} actualizado a ${nuevo_precio:.2f}"

        @self.server.tool(description="Lista todos los productos con filtro opcional de categoría")
        async def listar_productos(categoria: Optional[str] = None) -> str:
            conn = self._obtener_conexion()
            if categoria:
                cursor = conn.execute(
                    "SELECT id, nombre, precio FROM productos WHERE categoria = ?",
                    (categoria,)
                )
            else:
                cursor = conn.execute("SELECT id, nombre, precio FROM productos")

            rows = [dict(r) for r in cursor.fetchall()]
            return json.dumps(rows, ensure_ascii=False)

    def cerrar(self):
        if self._conn:
            self._conn.close()
            self._conn = None

    async def ejecutar_agente(self, prompt: str) -> str:
        options = ClaudeCodeOptions(
            mcp_servers=[self.server],
            max_turns=10
        )

        resultado = ""
        async for event in query(prompt=prompt, options=options):
            if hasattr(event, 'type') and event.type == 'result':
                resultado = event.result

        return resultado


# Uso
async def main():
    servidor = ServidorConEstado("/tmp/tienda.db")
    try:
        resultado = await servidor.ejecutar_agente(
            "¿Cuáles son los 3 productos más baratos de la categoría 'electrónica'?"
        )
        print(resultado)
    finally:
        servidor.cerrar()

---

Herramientas async y con estado

Herramientas que usan bases de datos con connection pooling

Para aplicaciones de producción, usar asyncpg con pool de conexiones:

import asyncio
import asyncpg
import json
from claude_code_sdk import query, ClaudeCodeOptions
from claude_code_sdk.mcp import create_sdk_mcp_server
from typing import Optional

class ServidorPostgresMCP:
    def __init__(self, dsn: str):
        self.dsn = dsn
        self.pool: Optional[asyncpg.Pool] = None
        self.server = create_sdk_mcp_server("postgres-tools")
        self._registrar_herramientas()

    async def inicializar(self):
        """Crear pool de conexiones antes de usar el servidor."""
        self.pool = await asyncpg.create_pool(
            self.dsn,
            min_size=2,
            max_size=10,
            command_timeout=30
        )
        print(f"Pool PostgreSQL inicializado ({self.pool.get_size()} conexiones)")

    async def cerrar(self):
        if self.pool:
            await self.pool.close()

    def _registrar_herramientas(self):
        @self.server.tool(description="Ejecuta una consulta SELECT en la base de datos")
        async def ejecutar_select(
            sql: str,
            parametros: Optional[list] = None
        ) -> str:
            # Validación de seguridad básica
            sql_upper = sql.strip().upper()
            if not sql_upper.startswith("SELECT"):
                return "Error: solo se permiten consultas SELECT"

            # Bloquear palabras clave peligrosas
            palabras_prohibidas = ["DROP", "DELETE", "INSERT", "UPDATE", "TRUNCATE", "ALTER"]
            for palabra in palabras_prohibidas:
                if palabra in sql_upper:
                    return f"Error: la consulta contiene la operación prohibida '{palabra}'"

            try:
                async with self.pool.acquire() as conn:
                    if parametros:
                        rows = await conn.fetch(sql, *parametros)
                    else:
                        rows = await conn.fetch(sql)

                    resultado = [dict(row) for row in rows]
                    return json.dumps(resultado, default=str, ensure_ascii=False)
            except asyncpg.PostgresError as e:
                return f"Error de base de datos: {str(e)}"

        @self.server.tool(description="Obtiene el conteo de registros en una tabla")
        async def contar_registros(
            tabla: str,
            condicion: Optional[str] = None
        ) -> str:
            # Validar nombre de tabla (solo alfanumérico y underscore)
            import re
            if not re.match(r'^[a-zA-Z_][a-zA-Z0-9_]*$', tabla):
                return f"Error: nombre de tabla inválido '{tabla}'"

            sql = f"SELECT COUNT(*) as total FROM {tabla}"
            if condicion:
                sql += f" WHERE {condicion}"

            try:
                async with self.pool.acquire() as conn:
                    row = await conn.fetchrow(sql)
                    return f"La tabla '{tabla}' tiene {row['total']} registros"
            except asyncpg.PostgresError as e:
                return f"Error: {str(e)}"

    async def ejecutar_agente(self, prompt: str) -> str:
        if not self.pool:
            raise RuntimeError("Llama a inicializar() antes de ejecutar_agente()")

        options = ClaudeCodeOptions(
            mcp_servers=[self.server],
            max_turns=15
        )

        resultado = ""
        async for event in query(prompt=prompt, options=options):
            if hasattr(event, 'type') and event.type == 'result':
                resultado = event.result
        return resultado


async def main():
    import os
    servidor = ServidorPostgresMCP(os.environ["DATABASE_URL"])
    await servidor.inicializar()
    try:
        resp = await servidor.ejecutar_agente(
            "¿Cuántos pedidos tiene cada estado? Muéstrame un resumen."
        )
        print(resp)
    finally:
        await servidor.cerrar()

asyncio.run(main())

Herramientas que llaman APIs externas con retry

import httpx
import asyncio
from claude_code_sdk.mcp import create_sdk_mcp_server
from typing import Optional
import json

server = create_sdk_mcp_server("api-externa")

async def llamar_api_con_retry(
    url: str,
    metodo: str = "GET",
    datos: Optional[dict] = None,
    max_intentos: int = 3,
    delay_base: float = 1.0
) -> dict:
    """Llama a una API con retry exponencial."""
    ultimo_error = None

    for intento in range(max_intentos):
        try:
            async with httpx.AsyncClient(timeout=30.0) as client:
                if metodo == "GET":
                    resp = await client.get(url)
                elif metodo == "POST":
                    resp = await client.post(url, json=datos)
                else:
                    raise ValueError(f"Método HTTP no soportado: {metodo}")

                resp.raise_for_status()
                return resp.json()

        except (httpx.TimeoutException, httpx.ConnectError) as e:
            ultimo_error = e
            if intento < max_intentos - 1:
                delay = delay_base * (2 ** intento)
                print(f"Intento {intento + 1} falló, esperando {delay}s...")
                await asyncio.sleep(delay)

        except httpx.HTTPStatusError as e:
            # No reintentar en errores 4xx (son errores del cliente)
            if 400 <= e.response.status_code < 500:
                raise
            ultimo_error = e
            if intento < max_intentos - 1:
                delay = delay_base * (2 ** intento)
                await asyncio.sleep(delay)

    raise ultimo_error


@server.tool(description="Obtiene el precio actual de una criptomoneda")
async def obtener_precio_cripto(simbolo: str) -> str:
    try:
        datos = await llamar_api_con_retry(
            f"https://api.coingecko.com/api/v3/simple/price"
            f"?ids={simbolo.lower()}&vs_currencies=usd,eur"
        )
        if simbolo.lower() not in datos:
            return f"Criptomoneda '{simbolo}' no encontrada"

        precios = datos[simbolo.lower()]
        return f"{simbolo.upper()}: USD ${precios.get('usd', 'N/A')} | EUR €{precios.get('eur', 'N/A')}"
    except Exception as e:
        return f"Error obteniendo precio: {str(e)}"

@server.tool(description="Verifica si un dominio está disponible para registro")
async def verificar_dominio(dominio: str) -> str:
    try:
        # WHOIS API (simplificado)
        datos = await llamar_api_con_retry(
            f"https://api.whoapi.com/?domain={dominio}&r=taken&apikey=DEMO"
        )
        disponible = datos.get("taken") == 0
        return f"El dominio '{dominio}' {'NO está disponible' if not disponible else 'ESTÁ disponible'}"
    except Exception as e:
        return f"No se pudo verificar '{dominio}': {str(e)}"

---

Combinar servidores in-process y externos

Ejemplo con 3 tipos de servidores distintos

import asyncio
import os
import sqlite3
import json
from claude_code_sdk import query, ClaudeCodeOptions, MCPServerStdio
from claude_code_sdk.mcp import create_sdk_mcp_server
from typing import Optional

# 1. Servidor in-process: lógica de negocio local
servidor_local = create_sdk_mcp_server("negocio")

@servidor_local.tool(description="Calcula el descuento aplicable a un pedido")
async def calcular_descuento(monto: float, cliente_premium: bool) -> str:
    if cliente_premium and monto > 1000:
        descuento = 0.20
    elif cliente_premium:
        descuento = 0.10
    elif monto > 500:
        descuento = 0.05
    else:
        descuento = 0.0

    monto_final = monto * (1 - descuento)
    return json.dumps({
        "monto_original": monto,
        "descuento_pct": descuento * 100,
        "monto_final": monto_final
    })

@servidor_local.tool(description="Valida si un número de tarjeta de crédito es válido (Luhn)")
async def validar_tarjeta(numero: str) -> str:
    numero_limpio = numero.replace(" ", "").replace("-", "")
    if not numero_limpio.isdigit():
        return "Inválida: contiene caracteres no numéricos"

    # Algoritmo de Luhn
    total = 0
    for i, digito in enumerate(reversed(numero_limpio)):
        n = int(digito)
        if i % 2 == 1:
            n *= 2
            if n > 9:
                n -= 9
        total += n

    valida = total % 10 == 0
    return f"Tarjeta {'VÁLIDA' if valida else 'INVÁLIDA'} (algoritmo Luhn)"


async def main():
    options = ClaudeCodeOptions(
        mcp_servers=[
            # Servidor 1: in-process (lógica de negocio)
            servidor_local,

            # Servidor 2: stdio (base de datos externa)
            MCPServerStdio(
                name="base_datos",
                command="npx",
                args=[
                    "-y", "@modelcontextprotocol/server-sqlite",
                    "--db-path", "/data/pedidos.db"
                ]
            ),

            # Servidor 3: HTTP (CRM externo)
            # MCPServerHTTP(
            #     name="crm",
            #     url="https://crm.empresa.com/mcp",
            #     headers={"Authorization": f"Bearer {os.environ['CRM_TOKEN']}"}
            # )
        ],
        max_turns=20,
        system_prompt="""Tienes acceso a:
        - mcp__negocio__*: lógica de negocio (descuentos, validaciones)
        - mcp__base_datos__*: base de datos de pedidos (SQLite)

        Usa estos servidores para responder preguntas sobre pedidos y validaciones."""
    )

    async for event in query(
        prompt="""Necesito procesar el pedido #1234:
        1. Busca el pedido en la base de datos
        2. Calcula el descuento (el cliente es premium con monto > $500)
        3. Valida la tarjeta terminada en 4532015112830366""",
        options=options
    ):
        if hasattr(event, 'type') and event.type == 'assistant':
            for block in event.message.content:
                if hasattr(block, 'text'):
                    print(block.text, end="")

asyncio.run(main())

Naming convention: mcp__server_name__tool_name

Cuando tienes múltiples servidores, las herramientas se nombran con este patrón:

mcp__{nombre_servidor}__{nombre_herramienta}

Por ejemplo:

• mcp__github__create_issue — herramienta create_issue del servidor github
• mcp__postgres__execute_query — herramienta execute_query del servidor postgres
• mcp__negocio__calcular_descuento — herramienta calcular_descuento del servidor negocio

allowed_tools para herramientas MCP

options = ClaudeCodeOptions(
    mcp_servers=[servidor_local, servidor_externo],

    # Solo permitir estas herramientas específicas de MCP
    allowed_tools=[
        "mcp__negocio__calcular_descuento",    # Solo esta del servidor local
        "mcp__base_datos__query",               # Solo query, no insert/delete
        "Read",                                  # Herramientas built-in también
        "Bash"
    ]
)

---

MCP con autenticación OAuth

Flujo OAuth para MCP servers

sequenceDiagram
    participant A as Agente
    participant MCP as MCP Client
    participant S as Servidor MCP
    participant AUTH as Auth Server (OAuth)

    A->>MCP: Iniciar sesión con servidor GitHub
    MCP->>S: initialize
    S-->>MCP: Requiere autenticación OAuth
    MCP->>AUTH: Solicitar token (client_credentials)
    AUTH-->>MCP: access_token + refresh_token
    MCP->>S: initialize + Bearer token
    S-->>MCP: Sesión establecida
    A->>MCP: Ejecutar herramienta
    MCP->>S: tools/call + Bearer token
    S-->>MCP: Resultado
    Note over MCP,AUTH: Token expira después de 1 hora
    MCP->>AUTH: Refresh con refresh_token
    AUTH-->>MCP: Nuevo access_token

Configuración con tokens OAuth

import asyncio
import time
import httpx
from dataclasses import dataclass
from typing import Optional
from claude_code_sdk import query, ClaudeCodeOptions, MCPServerHTTP

@dataclass
class TokenOAuth:
    access_token: str
    refresh_token: str
    expires_at: float

    def esta_expirado(self, buffer_segundos: int = 60) -> bool:
        return time.time() >= (self.expires_at - buffer_segundos)


class GestorTokenOAuth:
    def __init__(self, client_id: str, client_secret: str, token_url: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_url = token_url
        self._token: Optional[TokenOAuth] = None

    async def obtener_token(self) -> str:
        if self._token is None or self._token.esta_expirado():
            await self._renovar_token()
        return self._token.access_token

    async def _renovar_token(self):
        async with httpx.AsyncClient() as client:
            if self._token and self._token.refresh_token:
                # Usar refresh token
                data = {
                    "grant_type": "refresh_token",
                    "refresh_token": self._token.refresh_token,
                    "client_id": self.client_id,
                    "client_secret": self.client_secret
                }
            else:
                # Obtener token inicial
                data = {
                    "grant_type": "client_credentials",
                    "client_id": self.client_id,
                    "client_secret": self.client_secret,
                    "scope": "read:repos write:issues"
                }

            resp = await client.post(self.token_url, data=data)
            resp.raise_for_status()
            token_data = resp.json()

            self._token = TokenOAuth(
                access_token=token_data["access_token"],
                refresh_token=token_data.get("refresh_token", ""),
                expires_at=time.time() + token_data.get("expires_in", 3600)
            )


async def agente_con_oauth():
    import os
    gestor_token = GestorTokenOAuth(
        client_id=os.environ["OAUTH_CLIENT_ID"],
        client_secret=os.environ["OAUTH_CLIENT_SECRET"],
        token_url="https://github.com/login/oauth/access_token"
    )

    token = await gestor_token.obtener_token()

    options = ClaudeCodeOptions(
        mcp_servers=[
            MCPServerHTTP(
                name="github",
                url="https://api.githubcopilot.com/mcp/",
                headers={
                    "Authorization": f"Bearer {token}",
                    "Accept": "application/json"
                }
            )
        ]
    )

    async for event in query(
        prompt="Lista mis repositorios con más estrellas",
        options=options
    ):
        if hasattr(event, 'type') and event.type == 'result':
            print(event.result)

---

Ejemplo real: Agente con acceso a base de datos

Schema de la base de datos de ejemplo

-- Base de datos de e-commerce simplificada
CREATE TABLE usuarios (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    nombre TEXT NOT NULL,
    email TEXT UNIQUE NOT NULL,
    es_premium BOOLEAN DEFAULT FALSE,
    fecha_registro TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE productos (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    nombre TEXT NOT NULL,
    descripcion TEXT,
    precio REAL NOT NULL,
    stock INTEGER DEFAULT 0,
    categoria TEXT NOT NULL
);

CREATE TABLE pedidos (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    usuario_id INTEGER REFERENCES usuarios(id),
    total REAL NOT NULL,
    estado TEXT CHECK(estado IN ('pendiente', 'procesando', 'enviado', 'completado', 'cancelado')),
    fecha_pedido TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE items_pedido (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    pedido_id INTEGER REFERENCES pedidos(id),
    producto_id INTEGER REFERENCES productos(id),
    cantidad INTEGER NOT NULL,
    precio_unitario REAL NOT NULL
);

Agente CRUD completo con SQLite in-process

import asyncio
import sqlite3
import json
import re
from pathlib import Path
from claude_code_sdk import query, ClaudeCodeOptions
from claude_code_sdk.mcp import create_sdk_mcp_server
from typing import Optional, List

DB_PATH = "/tmp/ecommerce.db"

def inicializar_db():
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA foreign_keys = ON")
    # Crear schema...
    conn.execute("""
        CREATE TABLE IF NOT EXISTS productos (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            nombre TEXT NOT NULL,
            precio REAL NOT NULL,
            stock INTEGER DEFAULT 0,
            categoria TEXT NOT NULL
        )
    """)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS pedidos (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            usuario TEXT NOT NULL,
            total REAL NOT NULL,
            estado TEXT DEFAULT 'pendiente',
            fecha TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """)
    # Datos de prueba
    conn.execute("INSERT OR IGNORE INTO productos (nombre, precio, stock, categoria) VALUES ('Laptop Pro', 1299.99, 15, 'electronica')")
    conn.execute("INSERT OR IGNORE INTO productos (nombre, precio, stock, categoria) VALUES ('Mouse Inalámbrico', 49.99, 100, 'perifericos')")
    conn.execute("INSERT OR IGNORE INTO productos (nombre, precio, stock, categoria) VALUES ('Monitor 4K', 599.99, 8, 'electronica')")
    conn.commit()
    return conn


# Crear servidor con estado (conexión DB compartida)
conn_global = inicializar_db()
server = create_sdk_mcp_server("ecommerce-db")


@server.tool(description="Busca productos por nombre o categoría con filtros de precio")
async def buscar_productos(
    texto: Optional[str] = None,
    categoria: Optional[str] = None,
    precio_min: Optional[float] = None,
    precio_max: Optional[float] = None
) -> str:
    sql = "SELECT id, nombre, precio, stock, categoria FROM productos WHERE 1=1"
    params = []

    if texto:
        sql += " AND nombre LIKE ?"
        params.append(f"%{texto}%")
    if categoria:
        sql += " AND categoria = ?"
        params.append(categoria)
    if precio_min is not None:
        sql += " AND precio >= ?"
        params.append(precio_min)
    if precio_max is not None:
        sql += " AND precio <= ?"
        params.append(precio_max)

    cursor = conn_global.execute(sql, params)
    productos = [dict(row) for row in cursor.fetchall()]

    if not productos:
        return "No se encontraron productos con esos filtros"
    return json.dumps(productos, ensure_ascii=False)


@server.tool(description="Obtiene el detalle completo de un producto por ID")
async def obtener_producto(id_producto: int) -> str:
    cursor = conn_global.execute(
        "SELECT * FROM productos WHERE id = ?", (id_producto,)
    )
    row = cursor.fetchone()
    if not row:
        return f"Producto {id_producto} no encontrado"
    return json.dumps(dict(row), ensure_ascii=False)


@server.tool(description="Actualiza el precio de un producto")
async def actualizar_precio_producto(id_producto: int, nuevo_precio: float) -> str:
    if nuevo_precio <= 0:
        return "Error: el precio debe ser mayor a 0"

    cursor = conn_global.execute(
        "SELECT nombre FROM productos WHERE id = ?", (id_producto,)
    )
    if not cursor.fetchone():
        return f"Producto {id_producto} no encontrado"

    conn_global.execute(
        "UPDATE productos SET precio = ? WHERE id = ?",
        (nuevo_precio, id_producto)
    )
    conn_global.commit()
    return f"Precio del producto {id_producto} actualizado a ${nuevo_precio:.2f}"


@server.tool(description="Actualiza el stock de un producto (+ para agregar, - para reducir)")
async def actualizar_stock(id_producto: int, delta_stock: int) -> str:
    cursor = conn_global.execute(
        "SELECT nombre, stock FROM productos WHERE id = ?", (id_producto,)
    )
    row = cursor.fetchone()
    if not row:
        return f"Producto {id_producto} no encontrado"

    nuevo_stock = row["stock"] + delta_stock
    if nuevo_stock < 0:
        return f"Error: stock insuficiente. Stock actual: {row['stock']}, reducción solicitada: {abs(delta_stock)}"

    conn_global.execute(
        "UPDATE productos SET stock = ? WHERE id = ?",
        (nuevo_stock, id_producto)
    )
    conn_global.commit()
    return f"Stock de '{row['nombre']}' actualizado: {row['stock']} → {nuevo_stock}"


@server.tool(description="Crea un nuevo pedido")
async def crear_pedido(usuario: str, id_producto: int, cantidad: int) -> str:
    cursor = conn_global.execute(
        "SELECT nombre, precio, stock FROM productos WHERE id = ?", (id_producto,)
    )
    producto = cursor.fetchone()
    if not producto:
        return f"Producto {id_producto} no encontrado"

    if producto["stock"] < cantidad:
        return f"Stock insuficiente. Disponible: {producto['stock']}, solicitado: {cantidad}"

    total = producto["precio"] * cantidad

    conn_global.execute(
        "UPDATE productos SET stock = stock - ? WHERE id = ?",
        (cantidad, id_producto)
    )
    cursor = conn_global.execute(
        "INSERT INTO pedidos (usuario, total, estado) VALUES (?, ?, 'pendiente')",
        (usuario, total)
    )
    pedido_id = cursor.lastrowid
    conn_global.commit()

    return json.dumps({
        "pedido_id": pedido_id,
        "usuario": usuario,
        "producto": producto["nombre"],
        "cantidad": cantidad,
        "total": total,
        "estado": "pendiente"
    }, ensure_ascii=False)


@server.tool(description="Lista pedidos con filtro de estado opcional")
async def listar_pedidos(estado: Optional[str] = None, usuario: Optional[str] = None) -> str:
    sql = "SELECT * FROM pedidos WHERE 1=1"
    params = []
    if estado:
        sql += " AND estado = ?"
        params.append(estado)
    if usuario:
        sql += " AND usuario = ?"
        params.append(usuario)
    sql += " ORDER BY fecha DESC LIMIT 50"

    cursor = conn_global.execute(sql, params)
    pedidos = [dict(row) for row in cursor.fetchall()]
    return json.dumps(pedidos, ensure_ascii=False)


async def main():
    options = ClaudeCodeOptions(
        mcp_servers=[server],
        max_turns=15,
        system_prompt="Eres un asistente de e-commerce con acceso a la base de datos de productos y pedidos."
    )

    prompts = [
        "¿Cuántos productos de electrónica hay disponibles y cuál es el más caro?",
        "Crea un pedido para el usuario '[email protected]' de 2 unidades del Mouse Inalámbrico",
        "¿Cuánto vale el inventario total de electrónica? (precio * stock)"
    ]

    for prompt in prompts:
        print(f"\n{'='*60}")
        print(f"Consulta: {prompt}")
        print('='*60)
        async for event in query(prompt=prompt, options=options):
            if hasattr(event, 'type') and event.type == 'assistant':
                for block in event.message.content:
                    if hasattr(block, 'text'):
                        print(block.text, end="")
        print()

asyncio.run(main())

---

Ejemplo real: Agente de automatización web

Playwright MCP completo

import asyncio
import os
from claude_code_sdk import query, ClaudeCodeOptions, MCPServerStdio

async def agente_web_scraper():
    """Agente que extrae datos de páginas web usando Playwright."""
    options = ClaudeCodeOptions(
        mcp_servers=[
            MCPServerStdio(
                name="playwright",
                command="npx",
                args=[
                    "@playwright/mcp",
                    "--browser", "chromium",
                    "--headless"
                ],
                env={
                    "PATH": os.environ["PATH"],
                    "PLAYWRIGHT_BROWSERS_PATH": os.environ.get(
                        "PLAYWRIGHT_BROWSERS_PATH", ""
                    )
                }
            )
        ],
        max_turns=25,
        system_prompt="""Eres un agente de extracción de datos web.
        Usa Playwright para navegar páginas y extraer información estructurada.
        Siempre:
        1. Navega a la URL especificada
        2. Espera a que la página cargue completamente
        3. Extrae los datos solicitados
        4. Retorna los datos en formato JSON cuando sea posible"""
    )

    # Caso 1: Scraping de información de producto
    print("=== Extrayendo datos de producto ===")
    async for event in query(
        prompt="""Ve a https://books.toscrape.com/catalogue/a-light-in-the-attic_1000/index.html
        y extrae: título, precio, disponibilidad y descripción del libro.""",
        options=options
    ):
        if hasattr(event, 'type') and event.type == 'result':
            print(event.result)

async def agente_testing_ui():
    """Agente que hace testing de interfaz de usuario."""
    options = ClaudeCodeOptions(
        mcp_servers=[
            MCPServerStdio(
                name="playwright",
                command="npx",
                args=["@playwright/mcp", "--browser", "chromium"],
                env={"PATH": os.environ["PATH"]}
            )
        ],
        max_turns=30,
        system_prompt="""Eres un QA tester automatizado.
        Para cada tarea de testing:
        1. Navega al URL objetivo
        2. Ejecuta los pasos especificados
        3. Toma screenshots en momentos clave
        4. Verifica que los elementos esperados estén presentes
        5. Reporta cualquier anomalía encontrada"""
    )

    print("=== Testing de formulario de login ===")
    async for event in query(
        prompt="""Prueba el formulario de login en https://the-internet.herokuapp.com/login:
        1. Intenta login con credenciales incorrectas (user: 'wrong', pass: 'wrong')
        2. Verifica que aparece mensaje de error
        3. Intenta con credenciales correctas (user: 'tomsmith', pass: 'SuperSecretPassword!')
        4. Verifica que el login fue exitoso
        5. Toma screenshot del resultado final""",
        options=options
    ):
        if hasattr(event, 'type') and event.type == 'assistant':
            for block in event.message.content:
                if hasattr(block, 'text'):
                    print(block.text, end="")

asyncio.run(agente_web_scraper())

TypeScript equivalente:

import { query, ClaudeCodeOptions } from "@anthropic-ai/claude-code-sdk";

async function agenteWebScraper() {
  const options: ClaudeCodeOptions = {
    mcpServers: [{
      name: "playwright",
      command: "npx",
      args: ["@playwright/mcp", "--browser", "chromium", "--headless"],
      env: { PATH: process.env.PATH! }
    }],
    maxTurns: 25,
    systemPrompt: "Eres un agente de extracción de datos web con Playwright."
  };

  for await (const event of query({
    prompt: "Ve a https://example.com y extrae el título y todos los enlaces",
    options
  })) {
    if (event.type === "result") console.log(event.result);
  }
}

agenteWebScraper().catch(console.error);

---

Ejemplo real: Agente con GitHub

Flujo completo de code review automático

import asyncio
import os
from claude_code_sdk import query, ClaudeCodeOptions, MCPServerStdio

async def agente_code_review(repo: str, pr_number: int):
    """Agente que hace code review automático de un Pull Request."""
    options = ClaudeCodeOptions(
        mcp_servers=[
            MCPServerStdio(
                name="github",
                command="npx",
                args=["-y", "@modelcontextprotocol/server-github"],
                env={
                    "GITHUB_PERSONAL_ACCESS_TOKEN": os.environ["GITHUB_TOKEN"],
                    "PATH": os.environ["PATH"]
                }
            )
        ],
        max_turns=20,
        system_prompt=f"""Eres un revisor de código experto para el repositorio {repo}.

        Al revisar un PR:
        1. Obtén los archivos cambiados y su diff
        2. Revisa la lógica, seguridad, y estilo
        3. Identifica bugs potenciales
        4. Sugiere mejoras concretas
        5. Publica tu review con comentarios inline cuando sea posible

        Sé constructivo y específico. Proporciona ejemplos de código mejorado."""
    )

    prompt = f"""Haz un code review completo del PR #{pr_number} en {repo}.

    Estructura tu review así:
    1. **Resumen general**: ¿El PR cumple su propósito?
    2. **Problemas críticos** (bugs, seguridad): lista numerada
    3. **Mejoras sugeridas**: lista numerada
    4. **Aspectos positivos**: lo que está bien hecho
    5. **Veredicto**: APPROVE / REQUEST_CHANGES / COMMENT

    Publica los comentarios directamente en el PR."""

    resultado = []
    async for event in query(prompt=prompt, options=options):
        if hasattr(event, 'type'):
            if event.type == 'assistant':
                for block in event.message.content:
                    if hasattr(block, 'text'):
                        print(block.text, end="")
            elif event.type == 'result':
                resultado.append(event.result)

    return resultado


async def agente_ci_cd(repo: str, branch: str):
    """Agente que verifica el estado del CI/CD y actúa según el resultado."""
    options = ClaudeCodeOptions(
        mcp_servers=[
            MCPServerStdio(
                name="github",
                command="npx",
                args=["-y", "@modelcontextprotocol/server-github"],
                env={
                    "GITHUB_PERSONAL_ACCESS_TOKEN": os.environ["GITHUB_TOKEN"],
                    "PATH": os.environ["PATH"]
                }
            )
        ],
        max_turns=15
    )

    async for event in query(
        prompt=f"""Para el repositorio {repo} rama {branch}:
        1. Lista los últimos 5 workflow runs
        2. Identifica si hay alguno fallando
        3. Si hay fallas, analiza el log de errores y crea un issue descriptivo
        4. Si todo está bien, comenta en el último commit que el CI/CD está verde""",
        options=options
    ):
        if hasattr(event, 'type') and event.type == 'assistant':
            for block in event.message.content:
                if hasattr(block, 'text'):
                    print(block.text, end="")


async def main():
    repo = "owner/mi-repositorio"
    await agente_code_review(repo, pr_number=42)

asyncio.run(main())

---

Performance y optimización de MCP

Latencia in-process vs externo

flowchart LR
    subgraph LAT["Comparativa de latencia"]
        IP["In-Process\n~0.1ms"]
        STDIO["stdio (local)\n~2-10ms"]
        HTTP_L["HTTP (local)\n~5-20ms"]
        HTTP_R["HTTP (remoto)\n~50-500ms"]
    end

    IP -->|"x20 más lento"| STDIO
    STDIO -->|"x2 más lento"| HTTP_L
    HTTP_L -->|"x10-25 más lento"| HTTP_R

Pooling de conexiones para herramientas

from claude_code_sdk.mcp import create_sdk_mcp_server
import asyncio
from asyncio import Queue
import sqlite3

class PoolConexiones:
    def __init__(self, db_path: str, tamaño: int = 5):
        self.db_path = db_path
        self._pool: Queue = Queue(maxsize=tamaño)
        # Pre-crear conexiones
        for _ in range(tamaño):
            conn = sqlite3.connect(db_path, check_same_thread=False)
            conn.row_factory = sqlite3.Row
            self._pool.put_nowait(conn)

    async def adquirir(self) -> sqlite3.Connection:
        return await self._pool.get()

    def liberar(self, conn: sqlite3.Connection):
        self._pool.put_nowait(conn)


pool = PoolConexiones("/data/app.db", tamaño=5)
server = create_sdk_mcp_server("optimizado")

@server.tool(description="Query optimizado con pool de conexiones")
async def query_optimizado(sql: str) -> str:
    conn = await pool.adquirir()
    try:
        cursor = conn.execute(sql)
        rows = [dict(r) for r in cursor.fetchall()]
        return str(rows)
    finally:
        pool.liberar(conn)  # Siempre liberar, incluso en errores

---

Seguridad en herramientas MCP

Validar inputs y prevenir inyecciones

import re
import shlex
from claude_code_sdk.mcp import create_sdk_mcp_server

server = create_sdk_mcp_server("seguro")

# Lista blanca de tablas permitidas
TABLAS_PERMITIDAS = {"usuarios", "productos", "pedidos", "categorias"}

@server.tool(description="Consulta segura a la base de datos")
async def consulta_segura(tabla: str, condicion: Optional[str] = None) -> str:
    # 1. Validar nombre de tabla contra whitelist
    if tabla not in TABLAS_PERMITIDAS:
        return f"Error: tabla '{tabla}' no permitida. Tablas disponibles: {', '.join(TABLAS_PERMITIDAS)}"

    # 2. Si hay condición, validar que no contenga SQL injection
    if condicion:
        # Bloquear patrones peligrosos
        patrones_peligrosos = [
            r";\s*DROP", r";\s*DELETE", r";\s*UPDATE",
            r"--", r"/\*.*\*/", r"UNION\s+SELECT",
            r"OR\s+1\s*=\s*1", r"AND\s+1\s*=\s*1"
        ]
        for patron in patrones_peligrosos:
            if re.search(patron, condicion, re.IGNORECASE):
                return f"Error: condición contiene patrón SQL peligroso"

    sql = f"SELECT * FROM {tabla}"
    if condicion:
        sql += f" WHERE {condicion}"

    # Ejecutar consulta...
    return "Resultados..."

@server.tool(description="Ejecuta un comando del sistema (restringido)")
async def ejecutar_comando_seguro(comando: str) -> str:
    # Whitelist de comandos permitidos
    COMANDOS_PERMITIDOS = ["ls", "pwd", "date", "whoami", "df", "free"]

    # Parsear el comando y verificar el ejecutable
    try:
        partes = shlex.split(comando)
    except ValueError as e:
        return f"Error: comando malformado ({e})"

    if not partes:
        return "Error: comando vacío"

    ejecutable = partes[0]
    if ejecutable not in COMANDOS_PERMITIDOS:
        return f"Error: '{ejecutable}' no está permitido. Comandos disponibles: {', '.join(COMANDOS_PERMITIDOS)}"

    # Bloquear path traversal
    if ".." in comando or "/" in ejecutable:
        return "Error: path traversal no permitido"

    import subprocess
    try:
        resultado = subprocess.run(
            partes,
            capture_output=True,
            text=True,
            timeout=10  # Timeout de seguridad
        )
        return resultado.stdout or resultado.stderr
    except subprocess.TimeoutExpired:
        return "Error: el comando excedió el tiempo límite"

---

Crear un servidor MCP distribuible

Estructura de proyecto para publicar en npm

mi-servidor-mcp/
├── src/
│   ├── index.ts          # Entry point: crea y expone el servidor
│   ├── tools/
│   │   ├── consultas.ts  # Herramientas de consulta
│   │   └── mutaciones.ts # Herramientas de escritura
│   └── types.ts          # Tipos compartidos
├── package.json
├── tsconfig.json
└── README.md

package.json para publicar en npm:

{
  "name": "@miorg/mi-servidor-mcp",
  "version": "1.0.0",
  "description": "Servidor MCP para [descripción de tu integración]",
  "main": "dist/index.js",
  "bin": {
    "mi-servidor-mcp": "dist/cli.js"
  },
  "scripts": {
    "build": "tsc",
    "start": "node dist/cli.js",
    "prepublishOnly": "npm run build"
  },
  "keywords": ["mcp", "model-context-protocol", "claude"],
  "dependencies": {
    "@modelcontextprotocol/sdk": "^1.0.0"
  },
  "devDependencies": {
    "typescript": "^5.0.0"
  }
}

src/cli.ts — Wrapper CLI para uso con Claude Code:

#!/usr/bin/env node
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { crearServidor } from "./index.js";

async function main() {
  const servidor = crearServidor();
  const transport = new StdioServerTransport();
  await servidor.connect(transport);
  console.error("Servidor MCP iniciado y listo");
}

main().catch((error) => {
  console.error("Error fatal:", error);
  process.exit(1);
});

Configuración en settings.json de Claude Code:

{
  "mcpServers": {
    "mi-servidor": {
      "command": "npx",
      "args": ["-y", "@miorg/mi-servidor-mcp"],
      "env": {
        "API_KEY": "${MI_API_KEY}"
      }
    }
  }
}

Para publicar:

# Publicar en npm
npm login
npm publish --access public

# Publicar en PyPI (si es servidor Python)
pip install build twine
python -m build
twine upload dist/*

---

Resumen del capítulo

mindmap
  root((MCP))
    Protocolo
      Cliente-Servidor
      stdio
      HTTP/SSE
      JSON-RPC 2.0
    Servidores Oficiales
      Playwright
      GitHub
      PostgreSQL
      Filesystem
      Slack
      SQLite
    Tipos de Servidor
      Externo stdio
        Variables de entorno
        Subproceso
      Externo HTTP
        Bearer tokens
        OAuth
        SSE
      In-Process
        create_sdk_mcp_server
        @tool decorator
        Estado compartido
    Herramientas
      Tipos de parámetros
      Content types
        text
        image
        resource
      Manejo de errores
    Seguridad
      Whitelist herramientas
      SQL injection
      Command injection
      Validación inputs
    Optimización
      Latencia
      Connection pooling
      Cache