MCPExpert

Ejemplos y Casos de Uso

Explora ejemplos completos de servidores y clientes MCP, con código funcional y casos de uso reales.

Ejemplos de Servidores Completos

Servidores MCP funcionales listos para usar, con herramientas y recursos implementados.

Servidor de Archivos

Python

Servidor que expone archivos del sistema de archivos como recursos y permite leer/escribir archivos.

from mcp.server import Server
from mcp.types import Resource, Tool, TextContent
import mcp.types as types
import os

app = Server("file-server")

@app.list_resources()
async def list_resources() -> list[Resource]:
    """Lista archivos en un directorio"""
    resources = []
    base_path = os.getenv("DOCS_PATH", "./docs")
    
    for root, dirs, files in os.walk(base_path):
        for file in files:
            file_path = os.path.join(root, file)
            uri = f"file://{file_path}"
            resources.append(Resource(
                uri=uri,
                name=file,
                description=f"Archivo: {file}",
                mimeType="text/plain"
            ))
    return resources

@app.read_resource()
async def read_resource(uri: str) -> str:
    """Lee el contenido de un archivo"""
    if not uri.startswith("file://"):
        raise ValueError("URI inválida")
    
    file_path = uri.replace("file://", "")
    with open(file_path, "r", encoding="utf-8") as f:
        return f.read()

@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="write_file",
            description="Escribe contenido en un archivo",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {"type": "string"},
                    "content": {"type": "string"}
                },
                "required": ["path", "content"]
            }
        )
    ]

@app.call_tool("write_file")
async def write_file(path: str, content: str) -> TextContent:
    """Escribe contenido en un archivo"""
    with open(path, "w", encoding="utf-8") as f:
        f.write(content)
    return TextContent(text=f"Archivo {path} escrito exitosamente")

if __name__ == "__main__":
    from mcp.server.stdio import stdio_server
    stdio_server(app)
Incluye: recursos, herramientas, validación

Servidor SQLite

Python

Servidor que expone una base de datos SQLite como recursos y permite ejecutar consultas SQL.

from mcp.server import Server
from mcp.types import Resource, Tool, TextContent
import mcp.types as types
import sqlite3
import json

app = Server("sqlite-server")

DB_PATH = "example.db"

@app.list_resources()
async def list_resources() -> list[Resource]:
    """Lista todas las tablas como recursos"""
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
    tables = cursor.fetchall()
    conn.close()
    
    return [
        Resource(
            uri=f"db://table/{table[0]}",
            name=table[0],
            description=f"Tabla: {table[0]}",
            mimeType="application/json"
        )
        for table in tables
    ]

@app.read_resource()
async def read_resource(uri: str) -> str:
    """Lee el contenido de una tabla"""
    if not uri.startswith("db://table/"):
        raise ValueError("URI inválida")
    
    table_name = uri.replace("db://table/", "")
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    cursor.execute(f"SELECT * FROM {table_name}")
    rows = cursor.fetchall()
    columns = [desc[0] for desc in cursor.description]
    conn.close()
    
    data = [dict(zip(columns, row)) for row in rows]
    return json.dumps(data, indent=2)

@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="execute_query",
            description="Ejecuta una consulta SQL SELECT",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string"}
                },
                "required": ["query"]
            }
        )
    ]

@app.call_tool("execute_query")
async def execute_query(query: str) -> TextContent:
    """Ejecuta una consulta SQL"""
    if not query.strip().upper().startswith("SELECT"):
        raise ValueError("Solo se permiten consultas SELECT")
    
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    cursor.execute(query)
    rows = cursor.fetchall()
    columns = [desc[0] for desc in cursor.description]
    conn.close()
    
    result = [dict(zip(columns, row)) for row in rows]
    return TextContent(text=json.dumps(result, indent=2))

if __name__ == "__main__":
    from mcp.server.stdio import stdio_server
    stdio_server(app)
Incluye: recursos, herramientas, seguridad SQL

Servidor de API Externa

TypeScript

Servidor que integra una API externa (ej: GitHub) y expone datos como recursos y herramientas.

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
  CallToolRequestSchema,
  ListResourcesRequestSchema,
  ListToolsRequestSchema,
  ReadResourceRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";

const server = new Server(
  {
    name: "github-server",
    version: "0.1.0",
  },
  {
    capabilities: {
      resources: {},
      tools: {},
    },
  }
);

// Listar repositorios como recursos
server.setRequestHandler(ListResourcesRequestSchema, async () => {
  const response = await fetch(
    "https://api.github.com/users/octocat/repos",
    {
      headers: {
        Accept: "application/vnd.github.v3+json",
      },
    }
  );
  const repos = await response.json();
  
  return {
    resources: repos.map((repo: any) => ({
      uri: `github://repo/${repo.full_name}`,
      name: repo.name,
      description: repo.description || "",
      mimeType: "application/json",
    })),
  };
});

// Leer detalles de un repositorio
server.setRequestHandler(ReadResourceRequestSchema, async (request) => {
  const match = request.params.uri.match(/^github:\/\/repo\/(.+)$/);
  if (!match) {
    throw new Error("URI inválida");
  }
  
  const repoPath = match[1];
  const response = await fetch(
    `https://api.github.com/repos/${repoPath}`,
    {
      headers: {
        Accept: "application/vnd.github.v3+json",
      },
    }
  );
  const repo = await response.json();
  
  return {
    contents: [
      {
        uri: request.params.uri,
        mimeType: "application/json",
        text: JSON.stringify(repo, null, 2),
      },
    ],
  };
});

// Herramienta para buscar repositorios
server.setRequestHandler(ListToolsRequestSchema, async () => {
  return {
    tools: [
      {
        name: "search_repos",
        description: "Busca repositorios en GitHub",
        inputSchema: {
          type: "object",
          properties: {
            query: { type: "string" },
            language: { type: "string" },
          },
          required: ["query"],
        },
      },
    ],
  };
});

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  if (request.params.name === "search_repos") {
    const { query, language } = request.params.arguments as {
      query: string;
      language?: string;
    };
    
    let url = `https://api.github.com/search/repositories?q=${encodeURIComponent(query)}`;
    if (language) {
      url += `+language:${encodeURIComponent(language)}`;
    }
    
    const response = await fetch(url, {
      headers: {
        Accept: "application/vnd.github.v3+json",
      },
    });
    const data = await response.json();
    
    return {
      content: [
        {
          type: "text",
          text: JSON.stringify(data.items, null, 2),
        },
      ],
    };
  }
  
  throw new Error(`Herramienta desconocida: ${request.params.name}`);
});

async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.error("Servidor GitHub MCP iniciado");
}

main().catch(console.error);
Incluye: recursos, herramientas, integración API

Servidor de Mensajería

Python

Servidor que permite enviar mensajes a canales (Slack, Discord, etc.) y leer historial.

from mcp.server import Server
from mcp.types import Resource, Tool, TextContent
import mcp.types as types
import os
from datetime import datetime

app = Server("messaging-server")

# Simulación de almacenamiento de mensajes
messages_store = []

@app.list_resources()
async def list_resources() -> list[Resource]:
    """Lista canales disponibles como recursos"""
    channels = ["general", "dev", "support", "announcements"]
    return [
        Resource(
            uri=f"channel://{channel}",
            name=channel,
            description=f"Canal: {channel}",
            mimeType="application/json"
        )
        for channel in channels
    ]

@app.read_resource()
async def read_resource(uri: str) -> str:
    """Lee el historial de un canal"""
    if not uri.startswith("channel://"):
        raise ValueError("URI inválida")
    
    channel = uri.replace("channel://", "")
    channel_messages = [
        msg for msg in messages_store 
        if msg.get("channel") == channel
    ]
    
    import json
    return json.dumps(channel_messages, indent=2)

@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="send_message",
            description="Envía un mensaje a un canal",
            inputSchema={
                "type": "object",
                "properties": {
                    "channel": {"type": "string"},
                    "message": {"type": "string"},
                    "user": {"type": "string"}
                },
                "required": ["channel", "message"]
            }
        ),
        Tool(
            name="get_channel_stats",
            description="Obtiene estadísticas de un canal",
            inputSchema={
                "type": "object",
                "properties": {
                    "channel": {"type": "string"}
                },
                "required": ["channel"]
            }
        )
    ]

@app.call_tool("send_message")
async def send_message(channel: str, message: str, user: str = "bot") -> TextContent:
    """Envía un mensaje a un canal"""
    msg = {
        "channel": channel,
        "message": message,
        "user": user,
        "timestamp": datetime.now().isoformat()
    }
    messages_store.append(msg)
    return TextContent(text=f"Mensaje enviado a {channel}")

@app.call_tool("get_channel_stats")
async def get_channel_stats(channel: str) -> TextContent:
    """Obtiene estadísticas de un canal"""
    channel_messages = [
        msg for msg in messages_store 
        if msg.get("channel") == channel
    ]
    stats = {
        "channel": channel,
        "total_messages": len(channel_messages),
        "unique_users": len(set(msg.get("user") for msg in channel_messages))
    }
    import json
    return TextContent(text=json.dumps(stats, indent=2))

if __name__ == "__main__":
    from mcp.server.stdio import stdio_server
    stdio_server(app)
Incluye: recursos, herramientas, almacenamiento

Ejemplos de Clientes

Clientes MCP que se conectan a servidores y consumen sus recursos y herramientas.

Cliente Python

Python

Cliente completo que se conecta a un servidor MCP y consume recursos y herramientas.

import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def main():
    # Configurar conexión al servidor
    server_params = StdioServerParameters(
        command="python",
        args=["-m", "my_mcp_server"],
        env=None
    )
    
    # Conectar al servidor
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # Inicializar la sesión
            await session.initialize()
            
            # Listar recursos disponibles
            resources = await session.list_resources()
            print("Recursos disponibles:")
            for resource in resources.resources:
                print(f"  - {resource.name}: {resource.uri}")
            
            # Leer un recurso
            if resources.resources:
                first_resource = resources.resources[0]
                content = await session.read_resource(first_resource.uri)
                print(f"\nContenido de {first_resource.name}:")
                print(content.contents[0].text)
            
            # Listar herramientas disponibles
            tools = await session.list_tools()
            print("\nHerramientas disponibles:")
            for tool in tools.tools:
                print(f"  - {tool.name}: {tool.description}")
            
            # Llamar a una herramienta
            if tools.tools:
                first_tool = tools.tools[0]
                result = await session.call_tool(
                    first_tool.name,
                    arguments={}
                )
                print(f"\nResultado de {first_tool.name}:")
                print(result.content[0].text)

if __name__ == "__main__":
    asyncio.run(main())
Incluye: conexión, recursos, herramientas

Cliente TypeScript

TypeScript

Cliente TypeScript que se conecta a servidores MCP y consume sus capacidades.

import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";

async function main() {
  // Crear transporte stdio
  const transport = new StdioClientTransport({
    command: "node",
    args: ["dist/my-server.js"],
  });
  
  // Crear cliente
  const client = new Client(
    {
      name: "example-client",
      version: "0.1.0",
    },
    {
      capabilities: {},
    }
  );
  
  // Conectar al servidor
  await client.connect(transport);
  
  try {
    // Listar recursos
    const resources = await client.listResources();
    console.log("Recursos disponibles:");
    for (const resource of resources.resources) {
      console.log(`  - ${resource.name}: ${resource.uri}`);
    }
    
    // Leer un recurso
    if (resources.resources.length > 0) {
      const firstResource = resources.resources[0];
      const content = await client.readResource({
        uri: firstResource.uri,
      });
      console.log(`\nContenido de ${firstResource.name}:`);
      console.log(content.contents[0].text);
    }
    
    // Listar herramientas
    const tools = await client.listTools();
    console.log("\nHerramientas disponibles:");
    for (const tool of tools.tools) {
      console.log(`  - ${tool.name}: ${tool.description}`);
    }
    
    // Llamar a una herramienta
    if (tools.tools.length > 0) {
      const firstTool = tools.tools[0];
      const result = await client.callTool({
        name: firstTool.name,
        arguments: {},
      });
      console.log(`\nResultado de ${firstTool.name}:`);
      console.log(result.content[0].text);
    }
  } finally {
    await client.close();
  }
}

main().catch(console.error);
Incluye: conexión, recursos, herramientas

Casos de Uso Reales

Ejemplos prácticos de cómo MCP se utiliza en escenarios del mundo real.

Análisis de Código

Conecta un repositorio Git a Claude para análisis de código, refactoring y documentación automática.

Servidor:

Servidor que expone archivos de código como recursos y herramientas para análisis estático.

Herramientas:

  • Buscar funciones/clases
  • Analizar dependencias
  • Generar documentación

Beneficio:

Claude puede entender tu código base completo y sugerir mejoras específicas.

Integración CRM

Conecta Salesforce, HubSpot u otro CRM para que Claude pueda consultar y actualizar datos de clientes.

Servidor:

Servidor que expone contactos, oportunidades y actividades como recursos.

Herramientas:

  • Buscar contactos
  • Crear oportunidades
  • Registrar actividades

Beneficio:

Claude puede ayudar con gestión de relaciones con clientes de forma natural.

Documentación Interna

Expón documentación interna (Confluence, Notion, Wiki) para que Claude pueda responder preguntas específicas.

Servidor:

Servidor que indexa y expone documentos internos como recursos buscables.

Herramientas:

  • Buscar en documentación
  • Resumir artículos
  • Extraer información

Beneficio:

Claude tiene acceso a conocimiento interno de la empresa para respuestas precisas.

Monitoreo y Alertas

Conecta sistemas de monitoreo (Datadog, Prometheus) para que Claude pueda analizar métricas y alertas.

Servidor:

Servidor que expone métricas, logs y alertas como recursos.

Herramientas:

  • Consultar métricas
  • Analizar logs
  • Crear alertas

Beneficio:

Claude puede ayudar a diagnosticar problemas y optimizar sistemas.

Automatización de Tareas

Expón scripts y automatizaciones para que Claude pueda ejecutarlos y gestionar tareas repetitivas.

Servidor:

Servidor que expone scripts y comandos como herramientas ejecutables.

Herramientas:

  • Ejecutar scripts
  • Gestionar tareas
  • Programar trabajos

Beneficio:

Claude puede automatizar tareas complejas mediante comandos naturales.

Análisis de Datos

Conecta bases de datos y herramientas de BI para que Claude pueda realizar análisis y generar reportes.

Servidor:

Servidor que expone datos y consultas como recursos y herramientas.

Herramientas:

  • Ejecutar consultas SQL
  • Generar visualizaciones
  • Crear reportes

Beneficio:

Claude puede analizar datos complejos y responder preguntas de negocio.

¿Listo para crear tu propio servidor?

Revisa la documentación completa y los tutoriales para empezar a construir.