Discover Awesome MCP Servers

A2CR

MCP server for AI-agent handoffs with client-encrypted WorkBaton checkpoints and WorkStash notes.

Remote MCP with Azure Functions (Python)

A quickstart template to build and deploy a remote MCP server on Azure Functions using Python, secured with keys and HTTPS.

Template-Nodejs-MCP-Server

A Node.js template for building stateful MCP servers with streamable HTTP transport, featuring modular tools and resources for easy extension and integration.

Agentic Bits Claude Plugin

Adds a live git branch status bar to the Claude footer for tracking multiple active and reference repositories. It allows users to manage repository visibility and configuration directly through natural language MCP tools.

IL Bank MCP

Finance assistant that brings your Israeli bank data to any AI assistant, enabling transaction analysis, spending patterns, and financial insights.

3D Cartoon Generator & File System MCP Server

Enables high-quality 3D-style cartoon image generation using Google Gemini AI alongside secure file system operations like reading, writing, and managing directories. It provides a dual-purpose toolkit for creative asset generation and local file management via the Model Context Protocol.

LLM Inference Pricing Research Server

Enables scraping and comparing pricing information for LLM inference services across multiple providers (CloudRift, DeepInfra, Fireworks, Groq) using Firecrawl API and SQLite storage.

clipboard-vision-mcp

Enables text-only AI models to understand clipboard images by describing them through a vision model, eliminating manual file saving.

SUMO-MCP-Server

Enables AI agents to interact with Eclipse SUMO traffic simulations, automating tasks from data acquisition to signal optimization.

MCP Energy Server

Provides access to comprehensive U.S. and international energy data from the EIA API, including electricity, natural gas, petroleum, coal, renewables, CO2 emissions, and energy forecasts.

HUDU MCP Server

Enables interaction with HUDU technical documentation and customer management platform through API integration. Supports retrieving company information, searching knowledge base articles, and managing assets and credentials.

frogeye-mcp

AI-powered security vulnerability detection for MCP-compatible agents, offering real-time scanning of code against a knowledge graph of 24,000+ vulnerability patterns.

pay-mcp

USDC payments for AI agents on Base. Direct transfers, pre-funded tabs, x402 paywall handling, and service discovery.

ai-mcp-terminal

Multi-threaded terminal management MCP server for AI assistants, enabling async command execution, batch operations, and real-time web monitoring with up to 100 concurrent terminals.

emcp

MCP server providing filesystem operations, shell execution, and web search capabilities.

@qelos/better-mcp

A stdio MCP proxy that connects to one or more upstream MCP servers and exposes their tools, resources, and prompts through a single endpoint with a configurable middleware pipeline.

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.

OriginalVoices MCP Server

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

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.

CTF MCP Server

Exposes common CTF and cybersecurity tools (crypto, forensics, malware analysis, steganography, reverse engineering, pwn, OSINT) so LLMs can help solve capture-the-flag challenges in a controlled lab environment.

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.

mcp-abacus

Pure-Python MCP server for type-faithful calculation — evaluate expressions under fixed-point, IEEE-754 double, or exact rational arithmetic, with every answer labelled with its precision (exact vs inexact).

Lark MCP

Một máy chủ Giao thức Bối cảnh Mô hình (Model Context Protocol) cho phép các mô hình AI thực hiện các lệnh gọi hàm thông qua nền tảng nhắn tin Feishu/Lark, sử dụng tài khoản cá nhân của bạn (không cần cấu hình bot) để tạo ra một trợ lý AI đầy đủ tính năng.

mcp-open-webresearch

A proxy-aware MCP server that enables web searching across multiple engines and automated markdown content extraction from webpages. It features a deep research agent for recursive searching and synthesis, supporting complex network environments through SOCKS5 and HTTP proxies.

Date Operations MCP Server

Provides UK-centric date calculations including bank holiday integration, sprint planning tools, and specialized Asana workflow automation. It enables users to calculate working days, track upcoming holidays, and manage project schedules specifically within the Europe/London timezone.

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.

Rutherford MCP Server

Enables one AI coding agent to delegate tasks to, and build consensus across, multiple other coding CLIs (Claude Code, Codex, etc.) by orchestrating them as headless subprocesses.

iTerm2 Worktree MCP Server

Automates git worktree management with iTerm2 integration for Claude Code, allowing users to create, manage, and close git worktrees in isolated environments with automatic tab handling.