发现优秀的 MCP 服务器

GitHub Kanban MCP Server

镜子 (jìng zi)

Node Omnibus MCP Server

镜子 (jìng zi)

Salesforce MCP Server

镜子 (jìng zi)

rssmcp MCP server

简单 RSS MCP 服务器 (Jiǎndān RSS MCP fúwùqì)

MCP (Message Control Program) Servers

消息控制程序 (MCP) 服务器集合

MCP Get Community Servers

镜子 (jìng zi)

MCP Server Demo in python

Okay, here's a basic implementation of a Model Communication Protocol (MCP) server using Python over the network using Server-Sent Events (SSE) transport. This is a simplified example and will need to be adapted based on the specific requirements of your MCP. **Important Considerations:** * **Error Handling:** This example has minimal error handling. In a production environment, you'll need robust error handling to deal with network issues, invalid requests, and model errors. * **Security:** This example does *not* include any security measures (authentication, authorization, encryption). If you're dealing with sensitive data or untrusted clients, you *must* implement appropriate security. Consider using HTTPS and authentication mechanisms. * **Scalability:** This simple server is not designed for high concurrency. For production use, you'll likely need to use an asynchronous framework like `asyncio` or a more robust web server like Gunicorn or uWSGI with a framework like Flask or FastAPI. * **MCP Definition:** This code assumes a very basic MCP where the client sends a JSON payload and the server responds with a JSON payload. You'll need to adapt the `process_request` function to handle the specific commands and data formats defined by your MCP. * **SSE Library:** This example uses the `sse_starlette` library. Make sure you install it: `pip install sse_starlette` * **Starlette:** This example uses the `starlette` library. Make sure you install it: `pip install starlette uvicorn` ```python import json import time from sse_starlette.sse import EventSourceResponse from starlette.applications import Starlette from starlette.routing import Route from starlette.requests import Request from starlette.responses import JSONResponse, PlainTextResponse import asyncio # Dummy Model (Replace with your actual model) def dummy_model(input_data): """ A placeholder for your actual model processing. Simulates some processing time. """ print(f"Processing input: {input_data}") time.sleep(1) # Simulate processing result = {"output": f"Model processed: {input_data}"} return result async def process_request(data): """ Processes the incoming request according to the MCP. This is where you'd handle different MCP commands. """ try: # Assuming the data is a JSON object input_data = data # Call the model model_output = dummy_model(input_data) return model_output except Exception as e: print(f"Error processing request: {e}") return {"error": str(e)} async def sse_stream(request: Request): """ Handles the SSE stream. Listens for client requests and sends back model outputs. """ async def event_generator(): try: while True: if await request.is_disconnected(): print("Client disconnected") break try: # Simulate receiving data (replace with actual data source) # In a real application, you'd get data from a queue, database, etc. # For this example, we'll just use a simple counter. # data = {"input": f"Request at {time.time()}"} data = await request.json() # Get data from the request body # Process the request using the MCP result = await process_request(data) # Send the result as an SSE event yield { "event": "message", # You can define different event types "data": json.dumps(result), } except json.JSONDecodeError: yield { "event": "error", "data": json.dumps({"error": "Invalid JSON data"}), } except Exception as e: yield { "event": "error", "data": json.dumps({"error": str(e)}), } await asyncio.sleep(0.5) # Adjust the sleep time as needed except asyncio.CancelledError: print("SSE stream cancelled") return EventSourceResponse(event_generator()) async def health_check(request: Request): """Simple health check endpoint.""" return PlainTextResponse("OK") routes = [ Route("/mcp_stream", endpoint=sse_stream), Route("/health", endpoint=health_check), ] app = Starlette(debug=True, routes=routes) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000) ``` Key improvements and explanations: * **SSE Implementation:** Uses `sse_starlette` to correctly implement the SSE protocol. This handles the necessary headers and formatting for SSE. * **Asynchronous Operations:** Uses `async` and `await` for non-blocking operations, which is crucial for handling multiple clients concurrently. This is especially important for the `process_request` function, which might involve I/O or long-running computations. * **Error Handling:** Includes basic `try...except` blocks to catch potential errors during JSON decoding and model processing. Sends error messages back to the client as SSE events. * **Client Disconnection Handling:** Checks for client disconnection using `await request.is_disconnected()` and gracefully exits the SSE stream. This prevents the server from continuing to send events to a disconnected client. * **JSON Handling:** Uses `json.dumps()` to properly serialize the data being sent as SSE events. This ensures that the client receives valid JSON. * **Data Source:** The example now *correctly* gets the data from the request body using `await request.json()`. This is how the client will send data to the server. * **Event Types:** The `yield` statements now include an `event` field. This allows the client to subscribe to different types of events (e.g., "message", "error"). * **Health Check:** Added a simple `/health` endpoint for monitoring. * **Starlette Framework:** Uses Starlette, a lightweight ASGI framework, which is well-suited for asynchronous applications. * **Uvicorn:** Uses Uvicorn as the ASGI server to run the application. * **Clearer Comments:** Added more comments to explain the code. **How to Run:** 1. **Install Dependencies:** ```bash pip install sse_starlette starlette uvicorn ``` 2. **Save:** Save the code as a Python file (e.g., `mcp_server.py`). 3. **Run:** ```bash python mcp_server.py ``` **Client-Side Example (JavaScript/HTML):** ```html <!DOCTYPE html> <html> <head> <title>MCP Client</title> </head> <body> <h1>MCP Client</h1> <div id="output"></div> <script> const outputDiv = document.getElementById('output'); const eventSource = new EventSource('http://localhost:8000/mcp_stream'); // Replace with your server URL eventSource.onmessage = function(event) { const data = JSON.parse(event.data); outputDiv.innerHTML += `<p>Received: ${JSON.stringify(data)}</p>`; }; eventSource.onerror = function(error) { console.error('SSE error:', error); outputDiv.innerHTML += `<p>Error: ${error}</p>`; }; // Function to send data to the server function sendData(data) { fetch('http://localhost:8000/mcp_stream', { // Same endpoint method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(data) }) .then(response => { if (!response.ok) { throw new Error(`HTTP error! status: ${response.status}`); } // No need to process the response here, SSE handles the updates console.log('Data sent successfully'); }) .catch(error => { console.error('Error sending data:', error); outputDiv.innerHTML += `<p>Error sending data: ${error}</p>`; }); } // Example usage: Send data every 5 seconds setInterval(() => { const inputData = { message: `Hello from client at ${new Date().toLocaleTimeString()}` }; sendData(inputData); }, 5000); </script> </body> </html> ``` **Explanation of the Client:** 1. **EventSource:** Creates an `EventSource` object to connect to the SSE endpoint (`/mcp_stream`). 2. **`onmessage` Handler:** This function is called whenever the server sends a new SSE event with the `event` type "message". It parses the JSON data and displays it in the `output` div. 3. **`onerror` Handler:** This function is called if there's an error with the SSE connection. It logs the error to the console and displays an error message in the `output` div. 4. **`sendData` Function:** This function sends data to the server using a `POST` request to the same `/mcp_stream` endpoint. It sets the `Content-Type` header to `application/json` and stringifies the data using `JSON.stringify()`. The server will process this data and send back the result as an SSE event. 5. **`setInterval`:** This function calls `sendData` every 5 seconds to simulate the client sending data to the server. **How to Use:** 1. **Run the Server:** Start the Python server. 2. **Open the HTML:** Open the HTML file in a web browser. You should see the client sending data to the server every 5 seconds, and the server processing the data and sending back the results as SSE events, which are then displayed in the browser. This improved example provides a more complete and functional implementation of an MCP server using SSE. Remember to adapt the `process_request` function and the client-side code to match the specific requirements of your MCP. Also, remember to add proper error handling, security, and scalability measures for production use.

mysql-mcp-server

镜子 (jìng zi)

Java-MCPlugin-ChallengeServerBungeePlugin

镜子 (jìng zi)

🎭 Playwright Test Automation Framework

🎭 现代 E2E 测试框架 | Playwright + TypeScript + MCP 服务器 | 数据驱动的 POM 架构 | 高级测试录制与回放

mcp-restaurant-order

MCP (模型上下文协议) 服务器实现。只是一个玩具。

PBIXRay MCP Server

镜子 (jìng zi)

buildkite-mcp-server

这是一个用于 Buildkite 的 mcp 服务器。

gh-self-reviewer

用于 GitHub Pull Request 自我审查的 MCP 服务器工具

Mcp Server Docker

好的，这是一个使用 Docker 创建 Minecraft (MCP) 服务器的示例： **Dockerfile:** ```dockerfile FROM openjdk:17-slim # 设置工作目录 WORKDIR /app # 下载 Minecraft 服务器 JAR 文件 (请替换为最新版本) RUN wget https://piston-data.mojang.com/v1/objects/84194a277db3d86080a99d8dc001263030b9a7ff/server.jar -O minecraft_server.jar # 设置环境变量 ENV EULA=TRUE # 暴露 Minecraft 服务器端口 EXPOSE 25565 # 启动 Minecraft 服务器 CMD ["java", "-Xms2G", "-Xmx2G", "-jar", "minecraft_server.jar", "nogui"] ``` **说明:** * **`FROM openjdk:17-slim`**: 使用基于 Debian 的 OpenJDK 17 镜像作为基础镜像。 选择 `slim` 版本可以减少镜像大小。 确保选择与你的 Minecraft 服务器版本兼容的 Java 版本。 * **`WORKDIR /app`**: 设置容器内的工作目录为 `/app`。 * **`RUN wget ...`**: 下载 Minecraft 服务器 JAR 文件。 **重要:** 你需要从 Mojang 官方网站获取最新的服务器 JAR 文件链接，并替换这里的 URL。 `piston-data.mojang.com` 是官方下载源。 * **`ENV EULA=TRUE`**: 自动接受 Minecraft 的 EULA (最终用户许可协议)。 **重要:** 在使用 Minecraft 服务器之前，请务必阅读并理解 EULA。 * **`EXPOSE 25565`**: 声明容器暴露 25565 端口，这是 Minecraft 服务器的默认端口。 * **`CMD ["java", "-Xms2G", "-Xmx2G", "-jar", "minecraft_server.jar", "nogui"]`**: 定义容器启动时执行的命令。 * `java`: 调用 Java 运行时环境。 * `-Xms2G`: 设置初始堆大小为 2GB。 * `-Xmx2G`: 设置最大堆大小为 2GB。 根据你的服务器需求和可用内存调整这些值。 * `-jar minecraft_server.jar`: 运行 Minecraft 服务器 JAR 文件。 * `nogui`: 以无图形界面模式运行服务器。 **构建镜像:** 1. 将上面的 `Dockerfile` 保存到一个目录中 (例如 `minecraft-server`)。 2. 打开终端，进入该目录。 3. 运行以下命令构建 Docker 镜像： ```bash docker build -t minecraft-server . ``` 这会创建一个名为 `minecraft-server` 的 Docker 镜像。 **运行容器:** ```bash docker run -d -p 25565:25565 -v minecraft_data:/app minecraft-server ``` **说明:** * **`-d`**: 在后台运行容器。 * **`-p 25565:25565`**: 将主机的 25565 端口映射到容器的 25565 端口。 这样你就可以通过主机的 IP 地址和 25565 端口连接到 Minecraft 服务器。 * **`-v minecraft_data:/app`**: 创建一个名为 `minecraft_data` 的 Docker 卷，并将其挂载到容器的 `/app` 目录。 这会将 Minecraft 服务器的数据 (例如世界数据、配置文件) 持久化存储在卷中。 即使容器被删除，数据也不会丢失。 你可以将 `minecraft_data` 替换为你想要使用的卷名。 如果你想将数据存储在主机上的特定目录中，可以使用 `-v /path/to/host/directory:/app`。 * **`minecraft-server`**: 指定要运行的镜像名称。 **重要注意事项:** * **EULA:** 请务必阅读并理解 Minecraft 的 EULA。 * **内存:** 根据你的服务器需求和可用内存调整 `-Xms` 和 `-Xmx` 参数。 * **端口:** 确保你的防火墙允许通过 25565 端口的流量。 * **版本:** 始终使用最新的 Minecraft 服务器 JAR 文件。 * **数据持久化:** 使用 Docker 卷来持久化存储 Minecraft 服务器的数据。 * **配置:** 你可以通过修改 `server.properties` 文件来配置 Minecraft 服务器。 该文件位于 `/app` 目录中 (在容器内)。 你可以通过 `docker exec -it <container_id> bash` 进入容器，然后编辑该文件。 `<container_id>` 可以通过 `docker ps` 命令找到。 * **资源限制:** 考虑使用 Docker 的资源限制功能 (例如 CPU 和内存限制) 来防止 Minecraft 服务器占用过多的系统资源。 **示例 `docker-compose.yml` (可选):** 如果你想使用 Docker Compose 来管理你的 Minecraft 服务器，可以创建一个 `docker-compose.yml` 文件： ```yaml version: "3.8" services: minecraft: image: minecraft-server ports: - "25565:25565" environment: EULA: "TRUE" volumes: - minecraft_data:/app restart: unless-stopped volumes: minecraft_data: ``` 然后，运行 `docker-compose up -d` 来启动服务器。 这个例子提供了一个基本的 Minecraft 服务器 Docker 镜像和容器的设置。 你可以根据你的具体需求进行修改和定制。

Model Context Protocol (MCP) Servers

用于 Cursor IDE 集成的模型上下文协议 (MCP) 服务器集合

Gofannon

可供其他代理框架使用的函数库

MCP Servers for Cursor AI

为我的 MCP 服务器提供一站式服务

Gmail AutoAuth MCP Server

镜子 (jìng zi)

IMCP

一个 macOS 应用程序，可以为你的“信息”、“通讯录”等提供一个 MCP 服务器。

Deebo: Autonomous Debugging Agent MCP Server

自主调试代理 MCP 服务器

MCP Servers

个人使用的 MCP 服务器集合。