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
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)Servidor SQLite
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)Servidor de API Externa
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);Servidor de Mensajería
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)Ejemplos de Clientes
Clientes MCP que se conectan a servidores y consumen sus recursos y herramientas.
Cliente 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())Cliente 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);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.