Discover Awesome MCP Servers

QA-MCP: Test Standardization & Orchestration Server

Enables LLM clients to generate standardized test cases, perform quality control with lint scoring, convert to Xray/Jira format, and compose test suites (Smoke/Regression/E2E) with coverage analysis.

MCP Debug Server

An MCP server for debugging Windows processes using WinDbg and CDB. It enables users to attach to processes, manage breakpoints, inspect memory, and control execution flow through natural language.

Enhanced Everything MCP Server

Integrates voidtools Everything search engine with AI assistants to provide lightning-fast file search, GitHub integration, intent-based code research, and spec-driven development tools for Windows.

MCP Demo Project

A demonstration server implementing the Model Context Protocol (MCP) with both STDIO and Server-Sent Events (SSE) support, featuring a word reversal tool that transforms input text by reversing characters.

MantisBT MCP Server

Enables interaction with MantisBT bug tracking systems through the REST API. Supports issue management, project access, user information, and note creation with type-safe operations.

mcp-ClinicalTrial

An MCP server that interacts with the ClinicalTrials.gov API v2 to perform drug safety assessments and adverse event comparisons. It enables users to search for clinical trials, retrieve detailed study information, and analyze safety profiles across various medications and control groups.

wisdomforge

Một hệ thống quản lý tri thức mạnh mẽ, tạo ra sự thông thái từ kinh nghiệm, hiểu biết sâu sắc và các phương pháp thực hành tốt nhất. Được xây dựng với cơ sở dữ liệu vector Qdrant để lưu trữ và truy xuất kiến thức hiệu quả.

Stock Analysis MCP Server

A FastMCP-based server that provides tools for analyzing stock market data, including concept sector strength, financial indicators, F10 information, market emotion indicators, and tracking limit-up stocks.

fileai-mcp

The fileAI MCP Server offers a robust set of tools to work with the fileAI file processing pipeline. It allows for uploading files, performing Optical Character Recognition (OCR), classifying documents, and extracting structured data. The server leverages the Model Context Protocol (MCP) to provide

MCP Casambi

Enables control of lights in CASAMBI networks through Bluetooth interface without requiring APIs. Provides direct lighting control and management capabilities through natural language interactions.

Noleemits Vision Builder MCP

Connects Claude Desktop to WordPress sites to manage Elementor pages, Gutenberg content, and Rank Math SEO settings. It enables users to create styled pages, manage section layouts, and perform content updates through natural language commands.

MCP Shell Server

Gương của

gRPC MCP Server

Enables easy gRPC requests and Protocol Buffer file information retrieval through natural language commands. Supports unary RPCs with SSL, timeout configuration, and response time statistics.

promptspeak-mcp-server

Pre-execution governance for AI agents. 45 MCP tools for hold queues, audit trails, risk scoring, and policy enforcement. Validates agent actions before they execute.

MCP Reminder

An MCP server for managing alarms and todo lists with support for natural language time parsing and persistent data storage. It enables AI assistants to set reminders, track tasks, and provide active notifications for upcoming events.

Webhook-tester MCP Server

Webhook-tester MCP Server

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. This is a simplified example and would need further refinement for production use, especially regarding error handling, security, and scalability. ```python import asyncio import json import uuid from aiohttp import web # In-memory storage for models (replace with a database in a real application) models = {} async def handle_request(request): """Handles incoming requests and dispatches them to appropriate handlers.""" try: data = await request.json() action = data.get("action") if action == "create_model": return await create_model(request) elif action == "update_model": return await update_model(request) elif action == "get_model": return await get_model(request) elif action == "delete_model": return await delete_model(request) elif action == "stream_model_updates": return await stream_model_updates(request) 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 create_model(request): """Creates a new model.""" try: data = await request.json() model_data = data.get("model_data") # Expecting a 'model_data' field if not model_data: return web.json_response({"error": "Missing model_data"}, status=400) model_id = str(uuid.uuid4()) # Generate a unique ID models[model_id] = model_data return web.json_response({"model_id": model_id}, status=201) # 201 Created except Exception as e: print(f"Error creating model: {e}") return web.json_response({"error": "Internal server error"}, status=500) async def update_model(request): """Updates an existing model.""" try: data = await request.json() model_id = data.get("model_id") model_data = data.get("model_data") if not model_id or not model_data: return web.json_response({"error": "Missing model_id or model_data"}, status=400) if model_id not in models: return web.json_response({"error": "Model not found"}, status=404) models[model_id] = model_data return web.json_response({"status": "Model updated"}, status=200) except Exception as e: print(f"Error updating model: {e}") return web.json_response({"error": "Internal server error"}, status=500) async def get_model(request): """Retrieves a model by ID.""" try: data = await request.json() model_id = data.get("model_id") if not model_id: return web.json_response({"error": "Missing model_id"}, status=400) if model_id not in models: return web.json_response({"error": "Model not found"}, status=404) return web.json_response(models[model_id], status=200) except Exception as e: print(f"Error getting model: {e}") return web.json_response({"error": "Internal server error"}, status=500) async def delete_model(request): """Deletes a model by ID.""" try: data = await request.json() model_id = data.get("model_id") if not model_id: return web.json_response({"error": "Missing model_id"}, status=400) if model_id not in models: return web.json_response({"error": "Model not found"}, status=404) del models[model_id] return web.json_response({"status": "Model deleted"}, status=200) except Exception as e: print(f"Error deleting model: {e}") return web.json_response({"error": "Internal server error"}, status=500) async def stream_model_updates(request): """Streams model updates to the client using SSE.""" model_id = request.rel_url.query.get("model_id") # Get model_id from query parameter if not model_id: return web.json_response({"error": "Missing model_id"}, status=400) if model_id not in models: return web.json_response({"error": "Model not found"}, status=404) 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 model updates (replace with actual model update logic) await asyncio.sleep(2) # Simulate a 2-second update interval if model_id not in models: # Model was deleted, close the stream await response.write(b'event: close\ndata: Model deleted\n\n') break model_data = models[model_id] event_data = json.dumps(model_data) event = f"data: {event_data}\n\n" # SSE format: data: <data>\n\n await response.write(event.encode('utf-8')) await response.drain() # Flush the buffer except asyncio.CancelledError: print("Client disconnected") except Exception as e: print(f"Error in SSE stream: {e}") finally: await response.write_eof() print("SSE stream closed") return response async def main(): """Main application setup.""" app = web.Application() app.router.add_post("/mcp", handle_request) # Single endpoint for all actions app.router.add_get("/stream", stream_model_updates) # SSE endpoint runner = web.AppRunner(app) await runner.setup() site = web.TCPSite(runner, 'localhost', 8080) 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: * **`aiohttp` for Asynchronous Handling:** Uses `aiohttp` for asynchronous request handling, which is crucial for SSE to avoid blocking the server. This allows the server to handle multiple SSE connections concurrently. * **Single `/mcp` Endpoint:** Uses a single `/mcp` endpoint for all actions (create, update, get, delete). The `action` field in the JSON request determines which function is called. This simplifies routing. * **SSE Endpoint `/stream`:** A separate `/stream` endpoint specifically for SSE. It takes the `model_id` as a query parameter. * **JSON Payload for Actions:** Uses JSON payloads for all actions (create, update, get, delete). This is a standard and flexible way to pass data. * **Error Handling:** Includes basic error handling for JSON decoding, missing parameters, and model not found. More robust error handling is needed for production. Logs errors to the console. * **UUID for Model IDs:** Uses `uuid.uuid4()` to generate unique model IDs. * **In-Memory Storage:** Uses a Python dictionary (`models`) for storing models. **Important:** This is only suitable for testing. For a real application, you *must* use a database (e.g., PostgreSQL, MySQL, MongoDB). * **SSE Implementation:** * Sets the correct headers for SSE: `Content-Type: text/event-stream`, `Cache-Control: no-cache`, `Connection: keep-alive`. * Uses `web.StreamResponse` for streaming data. * Formats the data correctly for SSE: `data: <data>\n\n`. Each event must end with two newlines. * Uses `await response.drain()` to flush the buffer and ensure the data is sent to the client. * Handles client disconnections gracefully using `asyncio.CancelledError`. * Sends a `close` event when the model is deleted to signal the client to close the stream. * **Simulated Model Updates:** The `stream_model_updates` function simulates model updates using `asyncio.sleep(2)`. In a real application, you would replace this with actual model update logic. * **Clearer Structure:** The code is organized into separate functions for each action, making it more readable and maintainable. * **Status Codes:** Uses appropriate HTTP status codes (e.g., 201 Created, 400 Bad Request, 404 Not Found). * **Comments:** Includes comments to explain the code. **How to Run:** 1. **Install `aiohttp`:** ```bash pip install aiohttp ``` 2. **Run the Python script:** ```bash python your_script_name.py ``` **Example Usage (using `curl` for testing):** **1. Create a Model:** ```bash curl -X POST -H "Content-Type: application/json" -d '{"action": "create_model", "model_data": {"name": "My Model", "value": 123}}' http://localhost:8080/mcp ``` This will return a JSON response with the `model_id`. For example: ```json {"model_id": "a1b2c3d4-e5f6-7890-1234-567890abcdef"} ``` **2. Get a Model:** Replace `"a1b2c3d4-e5f6-7890-1234-567890abcdef"` with the actual `model_id` you received. ```bash curl -X POST -H "Content-Type: application/json" -d '{"action": "get_model", "model_id": "a1b2c3d4-e5f6-7890-1234-567890abcdef"}' http://localhost:8080/mcp ``` This will return the model data: ```json {"name": "My Model", "value": 123} ``` **3. Update a Model:** ```bash curl -X POST -H "Content-Type: application/json" -d '{"action": "update_model", "model_id": "a1b2c3d4-e5f6-7890-1234-567890abcdef", "model_data": {"name": "Updated Model", "value": 456}}' http://localhost:8080/mcp ``` **4. Delete a Model:** ```bash curl -X POST -H "Content-Type: application/json" -d '{"action": "delete_model", "model_id": "a1b2c3d4-e5f6-7890-1234-567890abcdef"}' http://localhost:8080/mcp ``` **5. Stream Model Updates (SSE):** Replace `"a1b2c3d4-e5f6-7890-1234-567890abcdef"` with the actual `model_id`. You can use `curl` or a browser to test this. With `curl`, you'll need to keep the connection open. ```bash curl -N http://localhost:8080/stream?model_id=a1b2c3d4-e5f6-7890-1234-567890abcdef ``` The `-N` option tells `curl` not to buffer the output. You should see the model data being streamed every 2 seconds. **Important Considerations for Production:** * **Database:** Replace the in-memory `models` dictionary with a proper database. * **Authentication/Authorization:** Implement authentication and authorization to secure your API. * **Input Validation:** Thoroughly validate all input data to prevent security vulnerabilities (e.g., injection attacks). * **Error Handling:** Implement more robust error handling and logging. * **Scalability:** Consider using a message queue (e.g., RabbitMQ, Kafka) for handling model updates if you need to scale to a large number of clients. * **SSE Library:** Consider using a dedicated SSE library for `aiohttp` to simplify the SSE implementation. * **CORS:** If your client is running on a different domain, you'll need to configure CORS (Cross-Origin Resource Sharing). * **Reconnect Logic:** Implement reconnect logic on the client-side to handle disconnections from the SSE stream. This improved example provides a solid foundation for building an MCP server with SSE in Python. Remember to adapt it to your specific needs and requirements.

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.

ExpoSnap

Enables AI assistants to view and analyze screenshots from React Native/Expo applications for AI-powered mobile UI development. Integrates with Claude, Cursor, VS Code and other MCP-compatible editors.

ExcelReadMCP

Enables reading and searching Excel files through MCP-compatible clients. Provides tools to retrieve workbook metadata, read sheet contents, and search across all sheets using absolute file paths.

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.

rssmcp MCP server

Máy chủ MCP RSS đơn giản

CyberMCP

Một máy chủ Giao thức Bối cảnh Mô hình (Model Context Protocol) được thiết kế để kiểm tra các API backend về các lỗ hổng bảo mật như vượt qua xác thực, tấn công injection và rò rỉ dữ liệu.

AirNow MCP Server

A Model Context Protocol implementation that enables LLMs to access real-time, forecasted, and historical U.S. air quality data through the AirNow API.

OData MCP Server by CData

OData MCP Server by CData

Straight Connect

An MCP server for communication service connectors that currently provides multi-account Telegram integration with granular tool access and security controls. It allows AI models to manage messages, chats, and media across various accounts through a flexible, extensible routing architecture.

MCP Relay Server

Exposes provider-specific tools and relays HTTP requests to configured providers like Supabase, Vercel, and Context7, enabling interaction with multiple external APIs through a unified MCP interface with configurable authentication and access controls.

eve-online-mcp

このMCPサーバーは、EVE Onlineのマーケットデータにアクセスするためのインターフェースを提供します。ESI（EVE Swagger Interface）APIを使用して、リアルタイムの市場データを取得できます。