Examples and Use Cases
Explore complete examples of MCP servers and clients, with functional code and real-world use cases.
Complete Server Examples
Functional MCP servers ready to use, with tools and resources implemented.
File Server
Server that exposes filesystem files as resources and allows reading/writing files.
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)SQLite Server
Server that exposes a SQLite database as resources and allows executing SQL queries.
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)External API Server
Server that integrates an external API (e.g., GitHub) and exposes data as resources and tools.
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);Messaging Server
Server that allows sending messages to channels (Slack, Discord, etc.) and reading history.
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)Client Examples
MCP clients that connect to servers and consume their resources and tools.
Python Client
Complete client that connects to an MCP server and consumes resources and tools.
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())TypeScript Client
TypeScript client that connects to MCP servers and consumes their capabilities.
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);Real-World Use Cases
Practical examples of how MCP is used in real-world scenarios.
Code Analysis
Connect a Git repository to Claude for code analysis, refactoring and automatic documentation.
Server:
Server that exposes code files as resources and tools for static analysis.
Tools:
- Search functions/classes
- Analyze dependencies
- Generate documentation
Benefit:
Claude can understand your entire codebase and suggest specific improvements.
CRM Integration
Connect Salesforce, HubSpot or another CRM so Claude can query and update customer data.
Server:
Server that exposes contacts, opportunities and activities as resources.
Tools:
- Search contacts
- Create opportunities
- Register activities
Benefit:
Claude can help with customer relationship management naturally.
Internal Documentation
Expose internal documentation (Confluence, Notion, Wiki) so Claude can answer specific questions.
Server:
Server that indexes and exposes internal documents as searchable resources.
Tools:
- Search documentation
- Summarize articles
- Extract information
Benefit:
Claude has access to internal company knowledge for accurate answers.
Monitoring and Alerts
Connect monitoring systems (Datadog, Prometheus) so Claude can analyze metrics and alerts.
Server:
Server that exposes metrics, logs and alerts as resources.
Tools:
- Query metrics
- Analyze logs
- Create alerts
Benefit:
Claude can help diagnose problems and optimize systems.
Task Automation
Expose scripts and automations so Claude can execute them and manage repetitive tasks.
Server:
Server that exposes scripts and commands as executable tools.
Tools:
- Execute scripts
- Manage tasks
- Schedule jobs
Benefit:
Claude can automate complex tasks through natural commands.
Data Analysis
Connect databases and BI tools so Claude can perform analysis and generate reports.
Server:
Server that exposes data and queries as resources and tools.
Tools:
- Execute SQL queries
- Generate visualizations
- Create reports
Benefit:
Claude can analyze complex data and answer business questions.
Ready to create your own server?
Check the complete documentation and tutorials to start building.