Discover Awesome MCP Servers

Crypto MCP Server

Provides real-time and historical cryptocurrency market data using ccxt, with tools for ticker, OHLCV, and streaming updates.

Spice MCP

Enables querying and analyzing blockchain data from Dune Analytics with Polars-optimized workflows, including schema discovery, Sui package exploration, and query management through natural language.

bilibili-mcp

A MCP tool to fetch Bilibili hot list videos, supporting configurable top-K results via async HTTP requests.

padlet-mcp

An MCP server that lets AI agents directly manage Padlet boards, including creating boards with sections, posting cards with attachments and polls, adding comments, reactions, and more.

Hello World MCP Server

A simple Model Context Protocol server that demonstrates basic functionality with greeting tools, allowing Claude to say hello and generate custom greetings with different styles and timestamps.

Portfolio Manager MCP Server

A Model Context Protocol server for managing and analyzing investment portfolios. It enables users to create and update portfolios, fetch real-time stock data and news, generate performance reports, and receive investment recommendations through natural language.

n8n MCP Server

Exposes n8n's REST API and local Ollama as MCP tools, allowing natural language instructions to create, execute, and monitor workflows from MCP clients like VS Code Copilot Chat.

AuraImage MCP Server

Enables AI agents to audit images for LCP savings, migrate assets to AuraImage CDN, generate alt text, create responsive tags, and preview smart crops directly from the editor.

safe-migrations-mcp

An MCP server that acts as a gatekeeper for database and configuration changes, requiring proposals, dry-run simulations, and confirmation tokens before applying any modifications.

tlaplus-mcp

Exposes the TLA+ toolchain (TLC, SANY, PlusCal, TLATeX) as structured JSON tools via the Model Context Protocol, enabling AI assistants to parse, check, simulate, and typeset TLA+ specifications.

Apple Notes MCP Server

Reading and writing Apple Notes with native Markdown conversion

x-ai-mcp

An MCP server that integrates X (Twitter) API access with Grok-powered intelligence for real-time social media analysis and account management. It provides tools for reading and writing tweets, managing direct messages, and generating AI-powered topic summaries or daily briefings.

CodeLogic

Interact with CodeLogic, a Software Intelligence platform that graphs complex code and data architecture dependencies, to boost AI accuracy and insight.

Cloudflare MCP Server

Enables AI assistants to manage Cloudflare resources through natural language, including DNS records, zone management, Workers KV storage, cache purging, and analytics. Supports comprehensive Cloudflare operations with secure API token authentication.

mcp-notas

Provides tools, resources, and prompts for managing a local Markdown notes database, including create, read, update, delete, search, and statistics operations, with robust path traversal protection.

midnight-nextjs-mcp

Combines Midnight Network blockchain development tools with Next.js DevTools for building decentralized applications, supporting contract development, wallet management, and Next.js diagnostics.

accessoticketing-mcp

An MCP server for accesso Passport mobile ticket links that reads orders, admissions, barcodes, and Google Wallet passes without login. It provides read-only tools to fetch ticket details, save barcode images, and resolve email tracking links.

Outlook MCP Server

Connects Claude to Microsoft Outlook through the Microsoft Graph API, enabling email management (list, search, read, send) and calendar operations (list, create, accept, decline, delete events) via OAuth 2.0 authentication.

entraadm-mcp

Enables read-only triage of Microsoft Entra ID sign-in and audit logs through fixed tools for sign-in failure analysis, user lookup, directory audits, and daily briefs.

Rope MCP

A Model Context Protocol server that exposes Python Rope refactoring capabilities to Claude Code, enabling safe symbol renaming, method extraction, code analysis, and more.

Lark MCP

Un servidor de Protocolo de Contexto de Modelo que permite a los modelos de IA realizar llamadas a funciones a través de la plataforma de mensajería Feishu/Lark, utilizando tu cuenta personal (sin necesidad de configuración de bot) para crear un asistente de IA con todas las funciones.

PixelForge MCP

PixelForge MCP provides a pipeline for converting AI-generated illustrations into in-game dot sprites. It automates background removal, pixelization, palette unification, outlining, and spritesheet packing via MCP tools.

Node Omnibus MCP Server

Espejo de

interactive-mcp

interactive-mcp

mcp-apps-ravenwatch

A stateless MCP server that generates interactive report widgets and survey builder UIs, using MCP Apps for rich visualizations and long-polling for live updates.

MCP Server Demo in python

Okay, here's a basic implementation of a Model Communication Protocol (MCP) server in Python using Server-Sent Events (SSE) for transport over a network. This is a simplified example to illustrate the core concepts. It will need further development for production use, including error handling, security, and more robust data serialization. ```python import asyncio import json import uuid from aiohttp import web # In-memory model registry (replace with a database in a real application) models = {} async def handle_request(request): """Handles incoming MCP requests.""" try: data = await request.json() action = data.get("action") if action == "register_model": return await register_model(data) elif action == "predict": return await predict(data) else: return web.json_response({"error": "Invalid action"}, status=400) except json.JSONDecodeError: return web.json_response({"error": "Invalid JSON"}, status=400) except Exception as e: print(f"Error processing request: {e}") # Log the error return web.json_response({"error": "Internal server error"}, status=500) async def register_model(data): """Registers a new model with the server.""" model_name = data.get("model_name") model_metadata = data.get("metadata", {}) # Allow optional metadata if not model_name: return web.json_response({"error": "Model name is required"}, status=400) model_id = str(uuid.uuid4()) # Generate a unique ID for the model models[model_id] = {"name": model_name, "metadata": model_metadata, "status": "ready"} # Store model info print(f"Registered model: {model_name} with ID: {model_id}") return web.json_response({"model_id": model_id, "message": "Model registered successfully"}, status=201) async def predict(data): """Handles prediction requests.""" model_id = data.get("model_id") input_data = data.get("input_data") if not model_id: return web.json_response({"error": "Model ID is required"}, status=400) if not input_data: return web.json_response({"error": "Input data is required"}, status=400) if model_id not in models: return web.json_response({"error": "Model not found"}, status=404) # Simulate model prediction (replace with actual model inference) await asyncio.sleep(0.1) # Simulate some processing time prediction_result = f"Prediction for model {model_id} with input: {input_data}" # Placeholder return web.json_response({"model_id": model_id, "prediction": prediction_result}) async def sse_handler(request): """Handles SSE connections and sends events.""" response = web.StreamResponse( status=200, headers={ 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', 'Connection': 'keep-alive', } ) await response.prepare(request) try: while True: # Simulate sending model status updates (replace with actual updates) for model_id, model_data in models.items(): event_data = { "model_id": model_id, "status": model_data["status"], "name": model_data["name"] } event = f"data: {json.dumps(event_data)}\n\n" # SSE format await response.write(event.encode('utf-8')) await response.drain() # Flush the buffer await asyncio.sleep(5) # Send updates every 5 seconds except asyncio.CancelledError: print("SSE connection closed") finally: await response.write_eof() return response async def health_check(request): """Simple health check endpoint.""" return web.json_response({"status": "ok"}) async def main(): """Main application setup.""" app = web.Application() app.add_routes([ web.post('/mcp', handle_request), # MCP endpoint web.get('/events', sse_handler), # SSE endpoint for model status web.get('/health', health_check), # Health check endpoint ]) runner = web.AppRunner(app) await runner.setup() site = web.TCPSite(runner, 'localhost', 8080) # Adjust host/port as needed await site.start() print("Server started at http://localhost:8080") try: await asyncio.Future() # Run forever except asyncio.CancelledError: print("Server shutting down...") finally: await runner.cleanup() if __name__ == "__main__": asyncio.run(main()) ``` Key improvements and explanations: * **Asynchronous using `aiohttp`:** Uses `aiohttp` for asynchronous request handling, which is crucial for scalability. This allows the server to handle many concurrent requests efficiently. * **SSE (Server-Sent Events):** Implements SSE for real-time updates to clients. The `sse_handler` function streams model status updates to connected clients. The `Content-Type` header is set correctly for SSE. `Cache-Control` is set to `no-cache` to prevent caching of the event stream. `Connection: keep-alive` helps maintain the connection. `response.drain()` is important to flush the buffer and ensure data is sent. * **MCP Structure:** The `handle_request` function parses the incoming JSON request and dispatches it to the appropriate handler based on the `action` field. This is the core of the MCP. * **Model Registry:** A simple in-memory `models` dictionary stores model information. In a real application, this would be replaced with a database (e.g., PostgreSQL, MongoDB). * **UUID for Model IDs:** Uses `uuid.uuid4()` to generate unique IDs for registered models. * **Error Handling:** Includes basic error handling for JSON decoding errors, missing parameters, and model not found. More robust error handling is needed for production. Errors are logged to the console. * **JSON Responses:** Uses `web.json_response` to create JSON responses with appropriate HTTP status codes. * **Health Check:** A simple `/health` endpoint is included for monitoring. * **Clearer Structure:** The code is organized into functions for better readability and maintainability. * **Comments:** Includes comments to explain the purpose of each section of the code. * **Simulated Prediction:** The `predict` function simulates model prediction with a `asyncio.sleep()` call. This should be replaced with actual model inference code. * **SSE Model Status Updates:** The `sse_handler` now sends model status updates every 5 seconds. This is a simulation; in a real application, you would trigger updates based on actual model status changes. The SSE data is formatted correctly as `data: {json}\n\n`. * **Graceful Shutdown:** Includes a `try...except asyncio.CancelledError...finally` block to handle server shutdown gracefully. * **`asyncio.run(main())`:** Uses `asyncio.run()` to start the asynchronous event loop. **How to Run:** 1. **Save:** Save the code as a Python file (e.g., `mcp_server.py`). 2. **Install Dependencies:** ```bash pip install aiohttp ``` 3. **Run:** ```bash python mcp_server.py ``` **Example Usage (Client-Side):** **Registering a Model (using `curl`):** ```bash curl -X POST -H "Content-Type: application/json" -d '{"action": "register_model", "model_name": "MyModel", "metadata": {"version": "1.0"}}' http://localhost:8080/mcp ``` **Making a Prediction (using `curl`):** ```bash curl -X POST -H "Content-Type: application/json" -d '{"action": "predict", "model_id": "YOUR_MODEL_ID", "input_data": {"feature1": 10, "feature2": 20}}' http://localhost:8080/mcp ``` Replace `YOUR_MODEL_ID` with the actual model ID returned from the registration request. **Connecting to the SSE Stream (using `curl`):** ```bash curl -N http://localhost:8080/events ``` The `-N` option tells `curl` not to buffer the output. You should see model status updates printed to the console every 5 seconds. **Important Considerations for Production:** * **Security:** Implement proper authentication and authorization to protect your models and data. Use HTTPS for secure communication. * **Model Loading and Management:** Develop a robust mechanism for loading, unloading, and managing models. Consider using a model serving framework like TensorFlow Serving, TorchServe, or ONNX Runtime. * **Scalability:** Use a load balancer to distribute traffic across multiple server instances. Consider using a message queue (e.g., RabbitMQ, Kafka) for asynchronous task processing. * **Monitoring and Logging:** Implement comprehensive monitoring and logging to track server performance and identify issues. * **Data Validation:** Validate incoming data to prevent errors and security vulnerabilities. * **Error Handling:** Implement more robust error handling and reporting. * **Serialization:** Choose an efficient serialization format (e.g., Protocol Buffers, Apache Arrow) for large data transfers. * **Database:** Use a proper database for storing model metadata and other persistent data. * **Asynchronous Tasks:** Offload long-running tasks (e.g., model loading, prediction) to background tasks using `asyncio.create_task()` or a task queue. * **SSE Reconnection:** Clients should implement reconnection logic for SSE in case the connection is lost. This improved example provides a solid foundation for building a more complete and robust MCP server. Remember to adapt it to your specific needs and requirements. **Spanish Translation of the Explanation:** Aquí tienes una implementación básica de un servidor de Protocolo de Comunicación de Modelos (MCP) en Python utilizando Eventos Enviados por el Servidor (SSE) para el transporte a través de una red. Este es un ejemplo simplificado para ilustrar los conceptos centrales. Necesitará un mayor desarrollo para su uso en producción, incluyendo el manejo de errores, la seguridad y una serialización de datos más robusta. ```python import asyncio import json import uuid from aiohttp import web # Registro de modelos en memoria (reemplazar con una base de datos en una aplicación real) models = {} async def handle_request(request): """Maneja las solicitudes MCP entrantes.""" try: data = await request.json() action = data.get("action") if action == "register_model": return await register_model(data) elif action == "predict": return await predict(data) else: return web.json_response({"error": "Acción inválida"}, status=400) except json.JSONDecodeError: return web.json_response({"error": "JSON inválido"}, status=400) except Exception as e: print(f"Error al procesar la solicitud: {e}") # Registra el error return web.json_response({"error": "Error interno del servidor"}, status=500) async def register_model(data): """Registra un nuevo modelo en el servidor.""" model_name = data.get("model_name") model_metadata = data.get("metadata", {}) # Permite metadatos opcionales if not model_name: return web.json_response({"error": "El nombre del modelo es obligatorio"}, status=400) model_id = str(uuid.uuid4()) # Genera un ID único para el modelo models[model_id] = {"name": model_name, "metadata": model_metadata, "status": "ready"} # Almacena la información del modelo print(f"Modelo registrado: {model_name} con ID: {model_id}") return web.json_response({"model_id": model_id, "message": "Modelo registrado con éxito"}, status=201) async def predict(data): """Maneja las solicitudes de predicción.""" model_id = data.get("model_id") input_data = data.get("input_data") if not model_id: return web.json_response({"error": "El ID del modelo es obligatorio"}, status=400) if not input_data: return web.json_response({"error": "Los datos de entrada son obligatorios"}, status=400) if model_id not in models: return web.json_response({"error": "Modelo no encontrado"}, status=404) # Simula la predicción del modelo (reemplazar con la inferencia real del modelo) await asyncio.sleep(0.1) # Simula un tiempo de procesamiento prediction_result = f"Predicción para el modelo {model_id} con entrada: {input_data}" # Marcador de posición return web.json_response({"model_id": model_id, "prediction": prediction_result}) async def sse_handler(request): """Maneja las conexiones SSE y envía eventos.""" response = web.StreamResponse( status=200, headers={ 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', 'Connection': 'keep-alive', } ) await response.prepare(request) try: while True: # Simula el envío de actualizaciones del estado del modelo (reemplazar con actualizaciones reales) for model_id, model_data in models.items(): event_data = { "model_id": model_id, "status": model_data["status"], "name": model_data["name"] } event = f"data: {json.dumps(event_data)}\n\n" # Formato SSE await response.write(event.encode('utf-8')) await response.drain() # Vacía el búfer await asyncio.sleep(5) # Envía actualizaciones cada 5 segundos except asyncio.CancelledError: print("Conexión SSE cerrada") finally: await response.write_eof() return response async def health_check(request): """Punto final simple de verificación de estado.""" return web.json_response({"status": "ok"}) async def main(): """Configuración principal de la aplicación.""" app = web.Application() app.add_routes([ web.post('/mcp', handle_request), # Punto final MCP web.get('/events', sse_handler), # Punto final SSE para el estado del modelo web.get('/health', health_check), # Punto final de verificación de estado ]) runner = web.AppRunner(app) await runner.setup() site = web.TCPSite(runner, 'localhost', 8080) # Ajusta el host/puerto según sea necesario await site.start() print("Servidor iniciado en http://localhost:8080") try: await asyncio.Future() # Ejecutar indefinidamente except asyncio.CancelledError: print("Servidor apagándose...") finally: await runner.cleanup() if __name__ == "__main__": asyncio.run(main()) ``` Mejoras y explicaciones clave: * **Asíncrono usando `aiohttp`:** Utiliza `aiohttp` para el manejo asíncrono de solicitudes, lo cual es crucial para la escalabilidad. Esto permite que el servidor maneje muchas solicitudes concurrentes de manera eficiente. * **SSE (Eventos Enviados por el Servidor):** Implementa SSE para actualizaciones en tiempo real a los clientes. La función `sse_handler` transmite actualizaciones del estado del modelo a los clientes conectados. El encabezado `Content-Type` se establece correctamente para SSE. `Cache-Control` se establece en `no-cache` para evitar el almacenamiento en caché del flujo de eventos. `Connection: keep-alive` ayuda a mantener la conexión. `response.drain()` es importante para vaciar el búfer y asegurar que los datos se envíen. * **Estructura MCP:** La función `handle_request` analiza la solicitud JSON entrante y la envía al manejador apropiado según el campo `action`. Este es el núcleo del MCP. * **Registro de Modelos:** Un simple diccionario `models` en memoria almacena la información del modelo. En una aplicación real, esto se reemplazaría con una base de datos (por ejemplo, PostgreSQL, MongoDB). * **UUID para IDs de Modelos:** Utiliza `uuid.uuid4()` para generar IDs únicos para los modelos registrados. * **Manejo de Errores:** Incluye un manejo de errores básico para errores de decodificación JSON, parámetros faltantes y modelo no encontrado. Se necesita un manejo de errores más robusto para la producción. Los errores se registran en la consola. * **Respuestas JSON:** Utiliza `web.json_response` para crear respuestas JSON con códigos de estado HTTP apropiados. * **Verificación de Estado:** Se incluye un punto final `/health` simple para la monitorización. * **Estructura Más Clara:** El código está organizado en funciones para una mejor legibilidad y mantenibilidad. * **Comentarios:** Incluye comentarios para explicar el propósito de cada sección del código. * **Predicción Simulada:** La función `predict` simula la predicción del modelo con una llamada a `asyncio.sleep()`. Esto debe reemplazarse con el código real de inferencia del modelo. * **Actualizaciones de Estado del Modelo SSE:** El `sse_handler` ahora envía actualizaciones del estado del modelo cada 5 segundos. Esto es una simulación; en una aplicación real, activarías las actualizaciones en función de los cambios reales del estado del modelo. Los datos SSE tienen el formato correcto como `data: {json}\n\n`. * **Apagado Grácil:** Incluye un bloque `try...except asyncio.CancelledError...finally` para manejar el apagado del servidor de forma grácil. * **`asyncio.run(main())`:** Utiliza `asyncio.run()` para iniciar el bucle de eventos asíncrono. **Cómo Ejecutar:** 1. **Guardar:** Guarda el código como un archivo Python (por ejemplo, `mcp_server.py`). 2. **Instalar Dependencias:** ```bash pip install aiohttp ``` 3. **Ejecutar:** ```bash python mcp_server.py ``` **Ejemplo de Uso (Lado del Cliente):** **Registrar un Modelo (usando `curl`):** ```bash curl -X POST -H "Content-Type: application/json" -d '{"action": "register_model", "model_name": "MyModel", "metadata": {"version": "1.0"}}' http://localhost:8080/mcp ``` **Realizar una Predicción (usando `curl`):** ```bash curl -X POST -H "Content-Type: application/json" -d '{"action": "predict", "model_id": "YOUR_MODEL_ID", "input_data": {"feature1": 10, "feature2": 20}}' http://localhost:8080/mcp ``` Reemplaza `YOUR_MODEL_ID` con el ID del modelo real devuelto por la solicitud de registro. **Conectarse al Flujo SSE (usando `curl`):** ```bash curl -N http://localhost:8080/events ``` La opción `-N` le dice a `curl` que no almacene en búfer la salida. Deberías ver las actualizaciones del estado del modelo impresas en la consola cada 5 segundos. **Consideraciones Importantes para la Producción:** * **Seguridad:** Implementa la autenticación y autorización adecuadas para proteger tus modelos y datos. Utiliza HTTPS para una comunicación segura. * **Carga y Gestión de Modelos:** Desarrolla un mecanismo robusto para cargar, descargar y gestionar modelos. Considera el uso de un marco de trabajo de servicio de modelos como TensorFlow Serving, TorchServe u ONNX Runtime. * **Escalabilidad:** Utiliza un balanceador de carga para distribuir el tráfico entre múltiples instancias del servidor. Considera el uso de una cola de mensajes (por ejemplo, RabbitMQ, Kafka) para el procesamiento asíncrono de tareas. * **Monitorización y Registro:** Implementa una monitorización y un registro exhaustivos para rastrear el rendimiento del servidor e identificar problemas. * **Validación de Datos:** Valida los datos entrantes para evitar errores y vulnerabilidades de seguridad. * **Manejo de Errores:** Implementa un manejo de errores e informes más robustos. * **Serialización:** Elige un formato de serialización eficiente (por ejemplo, Protocol Buffers, Apache Arrow) para grandes transferencias de datos. * **Base de Datos:** Utiliza una base de datos adecuada para almacenar los metadatos del modelo y otros datos persistentes. * **Tareas Asíncronas:** Descarga las tareas de larga duración (por ejemplo, la carga del modelo, la predicción) a tareas en segundo plano utilizando `asyncio.create_task()` o una cola de tareas. * **Reconexión SSE:** Los clientes deben implementar la lógica de reconexión para SSE en caso de que se pierda la conexión. Este ejemplo mejorado proporciona una base sólida para construir un servidor MCP más completo y robusto. Recuerda adaptarlo a tus necesidades y requisitos específicos.

Live Marketing Data MCP

Connect AI assistants to live Meta Ads, GA4, and Google Search Console data. 100% local, credentials machine-locked and encrypted. Supports Claude Desktop, Cursor, Windsurf, Cline, and more.

OriginalVoices MCP Server

Enables asking questions to AI twins representing specific audiences for research purposes.

Receita Federal NFS-e: Notas Emitidas

Enables querying issued electronic service invoices (NFS-e) from the Brazilian Federal Revenue via natural language, providing read-only access to official invoice data.

slug-generator

Enables AI agents to generate URL-friendly slugs from text with custom separators and transliteration, paid per call via x402 micropayments.