Discover Awesome MCP Servers

🦜 🔗 LangChain MCP Client

🦜🔗 LangChain Model Context Protocol (MCP) Client

MQTTX SSE Server

Una implementación del Protocolo Modelo-Contexto (MCP) que permite operaciones MQTT a través del transporte de Eventos Enviados por el Servidor (SSE).

HQ Pool Services Website

Test w/ Figma MCP server to generate a pool services page

🌟 Unsplash MCP Server Repository

🔎 A MCP server for Unsplash image search.

Trino MCP Server

Espejo de

Wisdom MCP Gateway

A stdio gateway for the Enterpret's Wisdom MCP SSE Server

Local iMessage RAG MCP Server

iMessage RAG MCP Server from Anthropic MCP Hackathon (NYC)

@enemyrr/mcp-mysql-server

Espejo de

MCP Actions Adapter

A simple adapter to convert a MCP server to a GPT actions compatible API

worker17

An MCP server to monitor workers productivity and fire them as needed.

Mcp Namecheap Registrar

Connects to namecheap api for checking availability and pricing of domains and registering them.

T2_C2

Server code for MCP

MCP Bundler Service

Un microservicio para empaquetar servidores MCP desde repositorios de GitHub y prepararlos para su despliegue.

Mcp Servers Collection

Colección de servidores e integraciones MCP verificados

mcp-server-yahoo-finance MCP server

Espejo de

MCP Test Client

MCP Test Client is a TypeScript testing utility for Model Context Protocol (MCP) servers.

ActionKit MCP Starter

Okay, here's a basic starter code structure and explanation for setting up an MCP (Minecraft Protocol) server powered by ActionKit (assuming you're referring to ActionKit, the open-source advocacy platform, and want to integrate it with a Minecraft server for some kind of interactive campaign or event). This is a conceptual outline, as the specific implementation will depend heavily on what you want to *do* with the integration. **Important Considerations:** * **ActionKit API:** You'll need to understand the ActionKit API to interact with it. This includes authentication, retrieving user data, and potentially creating actions (e.g., signing a petition, donating) based on in-game events. Refer to the ActionKit documentation for details. * **Minecraft Server API:** You'll need to use a Minecraft server API (e.g., Bukkit/Spigot, Fabric, or a lower-level library like `minecraft-protocol`) to interact with the Minecraft server. The choice depends on your needs and the level of control you require. Bukkit/Spigot are common for plugins, while Fabric is a modern alternative. `minecraft-protocol` gives you the most control but requires more coding. * **Language:** I'll assume you're using Python for the ActionKit interaction and Java/Kotlin for the Minecraft server plugin (if using Bukkit/Spigot or Fabric). You could use other languages, but these are common choices. * **Security:** Be extremely careful about security. Never expose your ActionKit API keys directly in your Minecraft plugin. Use environment variables or a secure configuration file. Validate all data received from the Minecraft server before sending it to ActionKit, and vice versa. * **Scalability:** Consider the scalability of your solution. If you expect a large number of players, you'll need to optimize your code and potentially use a message queue (e.g., RabbitMQ, Kafka) to handle the communication between the Minecraft server and ActionKit. **Conceptual Architecture:** 1. **Minecraft Server Plugin (Java/Kotlin):** * Listens for in-game events (e.g., player join, chat messages, specific actions). * Communicates with a separate Python script (or a web service) to interact with the ActionKit API. This communication can be done via HTTP requests (e.g., using `java.net.http` in Java 11+ or a library like OkHttp) or a message queue. * Receives data from the Python script (e.g., ActionKit user data, instructions) and uses it to modify the game world or player experience. 2. **Python Script (or Web Service):** * Receives requests from the Minecraft server plugin. * Authenticates with the ActionKit API. * Retrieves or creates data in ActionKit based on the requests. * Sends responses back to the Minecraft server plugin. **Starter Code (Conceptual):** **1. Minecraft Server Plugin (Java - Bukkit/Spigot Example):** ```java // ExamplePlugin.java package com.example; import org.bukkit.plugin.java.JavaPlugin; import org.bukkit.event.EventHandler; import org.bukkit.event.Listener; import org.bukkit.event.player.PlayerJoinEvent; import org.bukkit.entity.Player; import java.io.IOException; import java.net.URI; import java.net.http.HttpClient; import java.net.http.HttpRequest; import java.net.http.HttpResponse; import com.google.gson.Gson; import java.util.HashMap; import java.util.Map; public class ExamplePlugin extends JavaPlugin implements Listener { private final String ACTIONKIT_API_URL = "http://your-python-server:5000/actionkit"; // Replace with your Python server URL @Override public void onEnable() { getLogger().info("ExamplePlugin has been enabled!"); getServer().getPluginManager().registerEvents(this, this); } @Override public void onDisable() { getLogger().info("ExamplePlugin has been disabled!"); } @EventHandler public void onPlayerJoin(PlayerJoinEvent event) { Player player = event.getPlayer(); String playerName = player.getName(); // Asynchronously fetch ActionKit data for the player getServer().getScheduler().runTaskAsynchronously(this, () -> { try { String actionKitData = fetchActionKitData(playerName); // Process the ActionKit data (e.g., display a message, grant permissions) getServer().getScheduler().runTask(this, () -> { // Back to main thread for Bukkit API calls player.sendMessage("ActionKit Data: " + actionKitData); }); } catch (IOException | InterruptedException e) { getLogger().severe("Error fetching ActionKit data: " + e.getMessage()); getServer().getScheduler().runTask(this, () -> { player.sendMessage("Error fetching ActionKit data. See server logs."); }); } }); } private String fetchActionKitData(String playerName) throws IOException, InterruptedException { // Build the request to the Python server HttpClient client = HttpClient.newHttpClient(); Gson gson = new Gson(); Map<String, String> requestBodyMap = new HashMap<>(); requestBodyMap.put("username", playerName); String requestBody = gson.toJson(requestBodyMap); HttpRequest request = HttpRequest.newBuilder() .uri(URI.create(ACTIONKIT_API_URL)) .header("Content-Type", "application/json") .POST(HttpRequest.BodyPublishers.ofString(requestBody)) .build(); // Send the request and get the response HttpResponse<String> response = client.send(request, HttpResponse.BodyHandlers.ofString()); if (response.statusCode() == 200) { return response.body(); } else { getLogger().severe("ActionKit API request failed with status code: " + response.statusCode()); return "Error: " + response.statusCode(); } } } ``` **pom.xml (Maven - for building the plugin):** ```xml <project> <modelVersion>4.0.0</modelVersion> <groupId>com.example</groupId> <artifactId>ExamplePlugin</artifactId> <version>1.0-SNAPSHOT</version> <packaging>jar</packaging> <properties> <maven.compiler.source>17</maven.compiler.source> <maven.compiler.target>17</maven.compiler.target> </properties> <dependencies> <!-- Spigot API --> <dependency> <groupId>org.spigotmc</groupId> <artifactId>spigot-api</artifactId> <version>1.19.4-R0.1-SNAPSHOT</version> <!-- Replace with your Spigot version --> <scope>provided</scope> </dependency> <!-- Gson for JSON parsing --> <dependency> <groupId>com.google.code.gson</groupId> <artifactId>gson</artifactId> <version>2.8.9</version> </dependency> </dependencies> <build> <plugins> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-shade-plugin</artifactId> <version>3.2.4</version> <executions> <execution> <phase>package</phase> <goals> <goal>shade</goal> </goals> </execution> </executions> </plugin> </plugins> </build> </project> ``` **2. Python Script (Flask Example):** ```python # app.py from flask import Flask, request, jsonify import requests import os app = Flask(__name__) ACTIONKIT_API_BASE_URL = os.environ.get("ACTIONKIT_API_BASE_URL", "https://your-actionkit-instance.org/api/v1/") # Replace with your ActionKit URL ACTIONKIT_USERNAME = os.environ.get("ACTIONKIT_USERNAME", "your_username") ACTIONKIT_PASSWORD = os.environ.get("ACTIONKIT_PASSWORD", "your_password") def authenticate_actionkit(): """Authenticates with ActionKit and returns the session.""" session = requests.Session() session.auth = (ACTIONKIT_USERNAME, ACTIONKIT_PASSWORD) return session @app.route('/actionkit', methods=['POST']) def get_actionkit_data(): data = request.get_json() username = data.get('username') if not username: return jsonify({'error': 'Username is required'}), 400 try: session = authenticate_actionkit() # Example: Fetch user data from ActionKit (replace with your actual API call) user_url = f"{ACTIONKIT_API_BASE_URL}user/?username={username}" response = session.get(user_url) if response.status_code == 200: user_data = response.json() return jsonify(user_data) elif response.status_code == 404: return jsonify({'message': 'User not found in ActionKit'}), 404 else: print(f"ActionKit API Error: {response.status_code} - {response.text}") # Log the error return jsonify({'error': 'Error fetching data from ActionKit'}), 500 except Exception as e: print(f"Exception: {e}") # Log the exception return jsonify({'error': str(e)}), 500 if __name__ == '__main__': app.run(debug=True, host='0.0.0.0') # Make sure to set debug=False in production ``` **Explanation:** * **Minecraft Plugin (Java):** * `onPlayerJoin`: This event handler is triggered when a player joins the server. * `fetchActionKitData`: This method makes an HTTP POST request to the Python server, sending the player's username. It uses `java.net.http` (available in Java 11+) for making the HTTP request. It also uses Gson to serialize the request body to JSON. * The response from the Python server is then displayed to the player. * **Important:** The `runTaskAsynchronously` and `runTask` calls are crucial. Network requests should *always* be done asynchronously to avoid blocking the main server thread. Bukkit API calls (like `player.sendMessage`) *must* be done on the main thread. * **Python Script (Flask):** * `/actionkit` endpoint: This endpoint receives the username from the Minecraft plugin. * `authenticate_actionkit`: This function handles authentication with the ActionKit API using your username and password (stored as environment variables for security). * The script then makes a request to the ActionKit API to fetch user data based on the username. **Replace the example API call with the actual ActionKit API call you need.** * The response from ActionKit is then sent back to the Minecraft plugin as JSON. * **pom.xml:** This file defines the dependencies for your Java plugin (Spigot API and Gson). Make sure to update the Spigot API version to match your server. The `maven-shade-plugin` is used to create a single JAR file containing all the dependencies. **To Run This Example:** 1. **Set up ActionKit:** Make sure you have an ActionKit instance running and accessible. 2. **Set Environment Variables:** Set the `ACTIONKIT_API_BASE_URL`, `ACTIONKIT_USERNAME`, and `ACTIONKIT_PASSWORD` environment variables on the machine running the Python script. 3. **Install Python Dependencies:** `pip install flask requests` 4. **Run the Python Script:** `python app.py` 5. **Build the Minecraft Plugin:** Use Maven to build the plugin (e.g., `mvn clean install`). This will create a JAR file in the `target` directory. 6. **Install the Plugin:** Copy the JAR file to the `plugins` directory of your Spigot/Bukkit server. 7. **Start the Minecraft Server:** Start your Minecraft server. 8. **Join the Server:** Join the server with a Minecraft account. You should see a message in chat with the ActionKit data (or an error message if something went wrong). **Next Steps:** * **Implement Actual ActionKit Logic:** Replace the example ActionKit API call in the Python script with the actual API calls you need to retrieve or create data in ActionKit. * **Handle Different Events:** Add event handlers to the Minecraft plugin to listen for other in-game events (e.g., chat messages, player actions). * **Secure Communication:** Use HTTPS for communication between the Minecraft plugin and the Python server. * **Error Handling:** Implement robust error handling in both the Minecraft plugin and the Python script. Log errors to files or a logging service. * **Configuration:** Use a configuration file for the Minecraft plugin to store settings like the ActionKit API URL and other parameters. * **Message Queue (Optional):** If you need to handle a large number of requests, consider using a message queue (e.g., RabbitMQ, Kafka) to decouple the Minecraft plugin and the Python script. This is a basic starting point. You'll need to adapt it to your specific needs and requirements. Remember to prioritize security and scalability as you develop your integration. Good luck! **Spanish Translation of Key Points:** * **API de ActionKit:** Necesitarás entender la API de ActionKit para interactuar con ella. Esto incluye la autenticación, la recuperación de datos de usuario y, potencialmente, la creación de acciones (por ejemplo, firmar una petición, donar) basadas en eventos dentro del juego. Consulta la documentación de ActionKit para obtener más detalles. * **API del Servidor de Minecraft:** Necesitarás usar una API del servidor de Minecraft (por ejemplo, Bukkit/Spigot, Fabric o una biblioteca de nivel inferior como `minecraft-protocol`) para interactuar con el servidor de Minecraft. La elección depende de tus necesidades y del nivel de control que requieras. Bukkit/Spigot son comunes para los plugins, mientras que Fabric es una alternativa moderna. `minecraft-protocol` te da el mayor control, pero requiere más código. * **Lenguaje:** Asumiré que estás usando Python para la interacción con ActionKit y Java/Kotlin para el plugin del servidor de Minecraft (si usas Bukkit/Spigot o Fabric). Podrías usar otros lenguajes, pero estas son opciones comunes. * **Seguridad:** Ten mucho cuidado con la seguridad. Nunca expongas tus claves API de ActionKit directamente en tu plugin de Minecraft. Usa variables de entorno o un archivo de configuración seguro. Valida todos los datos recibidos del servidor de Minecraft antes de enviarlos a ActionKit, y viceversa. * **Escalabilidad:** Considera la escalabilidad de tu solución. Si esperas un gran número de jugadores, necesitarás optimizar tu código y, potencialmente, usar una cola de mensajes (por ejemplo, RabbitMQ, Kafka) para manejar la comunicación entre el servidor de Minecraft y ActionKit. The code comments are already in English, as that's the standard for code. If you need specific parts of the code translated, let me know.

ディーゼロ開発環境用 MCPサーバー

D-Zero frontend coding MCP server

Atlassian Jira MCP Server

Node.js/TypeScript MCP server for Atlassian Jira. Equips AI systems (LLMs) with tools to list/get projects, search/get issues (using JQL/ID), and view dev info (commits, PRs). Connects AI capabilities directly into Jira project management and issue tracking workflows.

Chrome MCP Server

Servidor MCP para integração entre extensão Chrome e Claude AI

mcp-server

Mirror of

📸 Smart Photo Journal MCP Server

Espejo de

MSSQL MCP Server

Espejo de

Weather App

Okay, here's a conceptual outline and code snippets illustrating a basic MCP (Microservice Communication Protocol) server implementation in Python for weather data, along with testing and pre-commit setup. This example focuses on clarity and demonstrates the core principles. It's a simplified illustration and would need further refinement for production use. **Conceptual Overview** 1. **MCP Server (Weather Service):** * Listens for requests on a defined port. * Receives MCP-formatted requests (e.g., for weather data at a specific location). * Processes the request. * Retrieves weather data (from a mock source in this example, but in a real application, this would be an external API or database). * Formats the response in MCP. * Sends the MCP response back to the client. 2. **MCP Format:** * We'll use a simple JSON-based MCP format for this example. A more robust implementation might use Protocol Buffers or other serialization methods. * Request: ```json { "method": "get_weather", "params": { "location": "London" }, "request_id": "unique_request_123" } ``` * Response: ```json { "result": { "temperature": 15, "condition": "Cloudy" }, "error": null, "request_id": "unique_request_123" } ``` or in case of error: ```json { "result": null, "error": { "code": -32601, "message": "Method not found" }, "request_id": "unique_request_123" } ``` 3. **Testing:** * Unit tests to verify the server's handling of different requests and error conditions. * Integration tests (optional) to test the interaction with external data sources (if applicable). 4. **Pre-commit Hooks:** * Automated checks (e.g., code formatting, linting, testing) that run before each commit to ensure code quality. **Code Snippets (Python)** ```python import socket import json import threading import time import uuid import logging import os # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') class WeatherService: def __init__(self, weather_data_source=None): self.weather_data = weather_data_source if weather_data_source else self._mock_weather_data() def _mock_weather_data(self): # In a real application, this would fetch data from an API or database. return { "London": {"temperature": 12, "condition": "Rainy"}, "Paris": {"temperature": 18, "condition": "Sunny"}, "Tokyo": {"temperature": 25, "condition": "Partly Cloudy"}, } def get_weather(self, location): if location in self.weather_data: return self.weather_data[location] else: return None class MCPRequestHandler: def __init__(self, client_socket, weather_service): self.client_socket = client_socket self.weather_service = weather_service def handle_request(self): try: data = self.client_socket.recv(1024).decode('utf-8') if not data: return # Connection closed logging.info(f"Received: {data}") request = json.loads(data) response = self.process_request(request) response_json = json.dumps(response) self.client_socket.sendall(response_json.encode('utf-8')) logging.info(f"Sent: {response_json}") except json.JSONDecodeError: error_response = self.create_error_response("Invalid JSON format", -32700, request_id=None) self.client_socket.sendall(json.dumps(error_response).encode('utf-8')) logging.error("Invalid JSON received") except Exception as e: logging.exception("Error processing request") error_response = self.create_error_response(f"Internal server error: {e}", -32603, request_id=request.get("request_id", None) if request else None) self.client_socket.sendall(json.dumps(error_response).encode('utf-8')) finally: self.client_socket.close() def process_request(self, request): method = request.get("method") params = request.get("params", {}) request_id = request.get("request_id") if not method: return self.create_error_response("Method not specified", -32600, request_id) if method == "get_weather": location = params.get("location") if not location: return self.create_error_response("Location not specified", -32602, request_id) weather_data = self.weather_service.get_weather(location) if weather_data: return self.create_success_response(weather_data, request_id) else: return self.create_error_response("Location not found", -32001, request_id) # Custom error code else: return self.create_error_response("Method not found", -32601, request_id) def create_success_response(self, result, request_id): return {"result": result, "error": None, "request_id": request_id} def create_error_response(self, message, code, request_id): return {"result": None, "error": {"code": code, "message": message}, "request_id": request_id} class MCPServer: def __init__(self, host, port, weather_service): self.host = host self.port = port self.weather_service = weather_service self.server_socket = None def start(self): self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) # Reuse address if server crashes self.server_socket.bind((self.host, self.port)) self.server_socket.listen(5) # Listen for up to 5 incoming connections logging.info(f"MCP Server listening on {self.host}:{self.port}") try: while True: client_socket, addr = self.server_socket.accept() logging.info(f"Accepted connection from {addr}") handler = MCPRequestHandler(client_socket, self.weather_service) threading.Thread(target=handler.handle_request).start() except KeyboardInterrupt: logging.info("Shutting down server...") finally: if self.server_socket: self.server_socket.close() logging.info("Server socket closed.") if __name__ == "__main__": HOST = "127.0.0.1" PORT = 65432 weather_service = WeatherService() server = MCPServer(HOST, PORT, weather_service) server.start() ``` **Example Client (for testing)** ```python import socket import json def send_mcp_request(host, port, method, params): with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.connect((host, port)) request = { "method": method, "params": params, "request_id": "test_request_" + str(time.time()) } request_json = json.dumps(request) s.sendall(request_json.encode('utf-8')) response_data = s.recv(1024).decode('utf-8') return json.loads(response_data) if __name__ == '__main__': HOST = "127.0.0.1" PORT = 65432 # Example usage response = send_mcp_request(HOST, PORT, "get_weather", {"location": "London"}) print(f"Response: {response}") response = send_mcp_request(HOST, PORT, "get_weather", {"location": "Atlantis"}) print(f"Response: {response}") response = send_mcp_request(HOST, PORT, "unknown_method", {"param1": "value1"}) print(f"Response: {response}") ``` **Testing (using `pytest`)** ```python import pytest import json from your_module import WeatherService, MCPRequestHandler # Replace your_module # Mock Weather Data for testing @pytest.fixture def mock_weather_data(): return { "TestCity": {"temperature": 20, "condition": "Sunny"}, "AnotherCity": {"temperature": 5, "condition": "Snowy"} } @pytest.fixture def weather_service(mock_weather_data): return WeatherService(mock_weather_data) def test_get_weather_success(weather_service): result = weather_service.get_weather("TestCity") assert result == {"temperature": 20, "condition": "Sunny"} def test_get_weather_not_found(weather_service): result = weather_service.get_weather("NonExistentCity") assert result is None def test_process_valid_request(weather_service): request_handler = MCPRequestHandler(None, weather_service) # No socket needed for this test request = { "method": "get_weather", "params": {"location": "TestCity"}, "request_id": "test_id" } response = request_handler.process_request(request) assert response["result"] == {"temperature": 20, "condition": "Sunny"} assert response["error"] is None assert response["request_id"] == "test_id" def test_process_invalid_method(weather_service): request_handler = MCPRequestHandler(None, weather_service) request = { "method": "unknown_method", "params": {}, "request_id": "test_id" } response = request_handler.process_request(request) assert response["result"] is None assert response["error"]["code"] == -32601 assert response["request_id"] == "test_id" def test_process_missing_location(weather_service): request_handler = MCPRequestHandler(None, weather_service) request = { "method": "get_weather", "params": {}, "request_id": "test_id" } response = request_handler.process_request(request) assert response["result"] is None assert response["error"]["code"] == -32602 assert response["request_id"] == "test_id" # Add more tests to cover error conditions, edge cases, etc. ``` **`pytest.ini` (for pytest configuration - optional)** ```ini [pytest] testpaths = . # Look for tests in the current directory python_files = test_*.py # Recognize files starting with "test_" as test files ``` **Pre-commit Setup** 1. **Install `pre-commit`:** ```bash pip install pre-commit ``` 2. **Create `.pre-commit-config.yaml`:** ```yaml repos: - repo: https://github.com/pre-commit/pre-commit-hooks rev: v4.5.0 # Use the latest stable version hooks: - id: trailing-whitespace - id: end-of-file-fixer - id: check-yaml - id: check-added-large-files - id: check-merge-conflict - repo: https://github.com/psf/black rev: 24.2.0 # Use the latest stable version hooks: - id: black - repo: https://github.com/PyCQA/flake8 rev: 7.0.0 # Use the latest stable version hooks: - id: flake8 args: ["--max-line-length=120"] # Adjust as needed ``` 3. **Install the pre-commit hooks:** ```bash pre-commit install ``` Now, before each `git commit`, the configured hooks will run. If any hook fails (e.g., `black` reformats your code, `flake8` finds linting errors), the commit will be aborted, and you'll need to fix the issues before committing again. **Explanation and Improvements** * **Error Handling:** The code includes basic error handling for JSON decoding, method not found, and missing parameters. More robust error handling would involve custom exception classes and more detailed error messages. * **Threading:** The server uses threads to handle multiple client connections concurrently. For high-performance applications, consider using asynchronous frameworks like `asyncio`. * **MCP Format:** The JSON-based MCP format is simple but can be extended to include versioning, metadata, and more complex data structures. Protocol Buffers are a popular choice for efficient serialization and schema definition. * **Data Source:** The `WeatherService` currently uses a mock data source. In a real application, you would integrate with a weather API (e.g., OpenWeatherMap, AccuWeather) or a database. * **Logging:** The code uses the `logging` module for basic logging. Configure logging levels and formats appropriately for your needs. * **Testing:** The `pytest` examples demonstrate unit testing of the `WeatherService` and `MCPRequestHandler`. Add more tests to cover all aspects of the server's functionality. Consider integration tests to verify the interaction with external data sources. * **Pre-commit:** The pre-commit configuration includes common hooks for code formatting (Black), linting (Flake8), and basic checks. Customize the hooks to match your project's requirements. * **Configuration:** Use environment variables or configuration files to manage settings like host, port, and API keys. * **Security:** For production environments, implement security measures such as authentication, authorization, and input validation to protect against malicious attacks. * **Request ID:** The `request_id` is crucial for correlating requests and responses, especially in asynchronous or distributed systems. It allows the client to match a response to the original request. The server should always include the `request_id` in the response. * **Error Codes:** Using standardized error codes (like those in JSON-RPC) makes it easier for clients to understand the nature of the error and handle it appropriately. Custom error codes can be used for application-specific errors. * **Graceful Shutdown:** Implement a mechanism for gracefully shutting down the server, allowing it to finish processing existing requests before exiting. This can be done using signals (e.g., `SIGTERM`). **How to Run** 1. Save the code snippets as separate Python files (e.g., `weather_server.py`, `weather_client.py`, `test_weather.py`). 2. Install the required packages: `pip install pytest pre-commit black flake8` 3. Run the server: `python weather_server.py` 4. Run the client in a separate terminal: `python weather_client.py` 5. Run the tests: `pytest` 6. Initialize pre-commit: `pre-commit install` This comprehensive example provides a solid foundation for building an MCP server for weather data. Remember to adapt and extend it to meet the specific requirements of your application. Spanish Translation: **Ejemplo de implementación de servidor MCP para datos meteorológicos con configuración de pruebas y pre-commit** Aquí tienes un esquema conceptual y fragmentos de código que ilustran una implementación básica de un servidor MCP (Protocolo de Comunicación de Microservicios) en Python para datos meteorológicos, junto con la configuración de pruebas y pre-commit. Este ejemplo se centra en la claridad y demuestra los principios básicos. Es una ilustración simplificada y necesitaría más refinamiento para su uso en producción. **Descripción general conceptual** 1. **Servidor MCP (Servicio Meteorológico):** * Escucha las solicitudes en un puerto definido. * Recibe solicitudes con formato MCP (por ejemplo, para datos meteorológicos en una ubicación específica). * Procesa la solicitud. * Recupera los datos meteorológicos (de una fuente simulada en este ejemplo, pero en una aplicación real, esto sería una API externa o una base de datos). * Formatea la respuesta en MCP. * Envía la respuesta MCP de vuelta al cliente. 2. **Formato MCP:** * Usaremos un formato MCP simple basado en JSON para este ejemplo. Una implementación más robusta podría usar Protocol Buffers u otros métodos de serialización. * Solicitud: ```json { "method": "get_weather", "params": { "location": "London" }, "request_id": "unique_request_123" } ``` * Respuesta: ```json { "result": { "temperature": 15, "condition": "Cloudy" }, "error": null, "request_id": "unique_request_123" } ``` o en caso de error: ```json { "result": null, "error": { "code": -32601, "message": "Method not found" }, "request_id": "unique_request_123" } ``` 3. **Pruebas:** * Pruebas unitarias para verificar el manejo del servidor de diferentes solicitudes y condiciones de error. * Pruebas de integración (opcional) para probar la interacción con fuentes de datos externas (si corresponde). 4. **Hooks de pre-commit:** * Comprobaciones automatizadas (por ejemplo, formateo de código, linting, pruebas) que se ejecutan antes de cada commit para garantizar la calidad del código. **Fragmentos de código (Python)** ```python import socket import json import threading import time import uuid import logging import os # Configurar el registro logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') class WeatherService: def __init__(self, weather_data_source=None): self.weather_data = weather_data_source if weather_data_source else self._mock_weather_data() def _mock_weather_data(self): # En una aplicación real, esto obtendría datos de una API o base de datos. return { "London": {"temperature": 12, "condition": "Rainy"}, "Paris": {"temperature": 18, "condition": "Sunny"}, "Tokyo": {"temperature": 25, "condition": "Partly Cloudy"}, } def get_weather(self, location): if location in self.weather_data: return self.weather_data[location] else: return None class MCPRequestHandler: def __init__(self, client_socket, weather_service): self.client_socket = client_socket self.weather_service = weather_service def handle_request(self): try: data = self.client_socket.recv(1024).decode('utf-8') if not data: return # Conexión cerrada logging.info(f"Recibido: {data}") request = json.loads(data) response = self.process_request(request) response_json = json.dumps(response) self.client_socket.sendall(response_json.encode('utf-8')) logging.info(f"Enviado: {response_json}") except json.JSONDecodeError: error_response = self.create_error_response("Formato JSON inválido", -32700, request_id=None) self.client_socket.sendall(json.dumps(error_response).encode('utf-8')) logging.error("JSON inválido recibido") except Exception as e: logging.exception("Error al procesar la solicitud") error_response = self.create_error_response(f"Error interno del servidor: {e}", -32603, request_id=request.get("request_id", None) if request else None) self.client_socket.sendall(json.dumps(error_response).encode('utf-8')) finally: self.client_socket.close() def process_request(self, request): method = request.get("method") params = request.get("params", {}) request_id = request.get("request_id") if not method: return self.create_error_response("Método no especificado", -32600, request_id) if method == "get_weather": location = params.get("location") if not location: return self.create_error_response("Ubicación no especificada", -32602, request_id) weather_data = self.weather_service.get_weather(location) if weather_data: return self.create_success_response(weather_data, request_id) else: return self.create_error_response("Ubicación no encontrada", -32001, request_id) # Código de error personalizado else: return self.create_error_response("Método no encontrado", -32601, request_id) def create_success_response(self, result, request_id): return {"result": result, "error": None, "request_id": request_id} def create_error_response(self, message, code, request_id): return {"result": None, "error": {"code": code, "message": message}, "request_id": request_id} class MCPServer: def __init__(self, host, port, weather_service): self.host = host self.port = port self.weather_service = weather_service self.server_socket = None def start(self): self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) # Reutilizar la dirección si el servidor falla self.server_socket.bind((self.host, self.port)) self.server_socket.listen(5) # Escuchar hasta 5 conexiones entrantes logging.info(f"Servidor MCP escuchando en {self.host}:{self.port}") try: while True: client_socket, addr = self.server_socket.accept() logging.info(f"Conexión aceptada desde {addr}") handler = MCPRequestHandler(client_socket, self.weather_service) threading.Thread(target=handler.handle_request).start() except KeyboardInterrupt: logging.info("Apagando el servidor...") finally: if self.server_socket: self.server_socket.close() logging.info("Socket del servidor cerrado.") if __name__ == "__main__": HOST = "127.0.0.1" PORT = 65432 weather_service = WeatherService() server = MCPServer(HOST, PORT, weather_service) server.start() ``` **Ejemplo de cliente (para pruebas)** ```python import socket import json def send_mcp_request(host, port, method, params): with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.connect((host, port)) request = { "method": method, "params": params, "request_id": "test_request_" + str(time.time()) } request_json = json.dumps(request) s.sendall(request_json.encode('utf-8')) response_data = s.recv(1024).decode('utf-8') return json.loads(response_data) if __name__ == '__main__': HOST = "127.0.0.1" PORT = 65432 # Ejemplo de uso response = send_mcp_request(HOST, PORT, "get_weather", {"location": "London"}) print(f"Respuesta: {response}") response = send_mcp_request(HOST, PORT, "get_weather", {"location": "Atlantis"}) print(f"Respuesta: {response}") response = send_mcp_request(HOST, PORT, "unknown_method", {"param1": "value1"}) print(f"Respuesta: {response}") ``` **Pruebas (usando `pytest`)** ```python import pytest import json from your_module import WeatherService, MCPRequestHandler # Reemplaza your_module # Datos meteorológicos simulados para pruebas @pytest.fixture def mock_weather_data(): return { "TestCity": {"temperature": 20, "condition": "Sunny"}, "AnotherCity": {"temperature": 5, "condition": "Snowy"} } @pytest.fixture def weather_service(mock_weather_data): return WeatherService(mock_weather_data) def test_get_weather_success(weather_service): result = weather_service.get_weather("TestCity") assert result == {"temperature": 20, "condition": "Sunny"} def test_get_weather_not_found(weather_service): result = weather_service.get_weather("NonExistentCity") assert result is None def test_process_valid_request(weather_service): request_handler = MCPRequestHandler(None, weather_service) # No se necesita socket para esta prueba request = { "method": "get_weather", "params": {"location": "TestCity"}, "request_id": "test_id" } response = request_handler.process_request(request) assert response["result"] == {"temperature": 20, "condition": "Sunny"} assert response["error"] is None assert response["request_id"] == "test_id" def test_process_invalid_method(weather_service): request_handler = MCPRequestHandler(None, weather_service) request = { "method": "unknown_method", "params": {}, "request_id": "test_id" } response = request_handler.process_request(request) assert response["result"] is None assert response["error"]["code"] == -32601 assert response["request_id"] == "test_id" def test_process_missing_location(weather_service): request_handler = MCPRequestHandler(None, weather_service) request = { "method": "get_weather", "params": {}, "request_id": "test_id" } response = request_handler.process_request(request) assert response["result"] is None assert response["error"]["code"] == -32602 assert response["request_id"] == "test_id" # Agrega más pruebas para cubrir condiciones de error, casos límite, etc. ``` **`pytest.ini` (para la configuración de pytest - opcional)** ```ini [pytest] testpaths = . # Busca pruebas en el directorio actual python_files = test_*.py # Reconoce los archivos que comienzan con "test_" como archivos de prueba ``` **Configuración de pre-commit** 1. **Instala `pre-commit`:** ```bash pip install pre-commit ``` 2. **Crea `.pre-commit-config.yaml`:** ```yaml repos: - repo: https://github.com/pre-commit/pre-commit-hooks rev: v4.5.0 # Usa la última versión estable hooks: - id: trailing-whitespace - id: end-of-file-fixer - id: check-yaml - id: check-added-large-files - id: check-merge-conflict - repo: https://github.com/psf/black rev: 24.2.0 # Usa la última versión estable hooks: - id: black - repo: https://github.com/PyCQA/flake8 rev: 7.0.0 # Usa la última versión estable hooks: - id: flake8 args: ["--max-line-length=120"] # Ajusta según sea necesario ``` 3. **Instala los hooks de pre-commit:** ```bash pre-commit install ``` Ahora, antes de cada `git commit`, se ejecutarán los hooks configurados. Si algún hook falla (por ejemplo, `black` reformatea tu código, `flake8` encuentra errores de linting), el commit se abortará y deberás solucionar los problemas antes de volver a hacer commit. **Explicación y mejoras** * **Manejo de errores:** El código incluye un manejo de errores básico para la decodificación de JSON, el método no encontrado y los parámetros faltantes. Un manejo de errores más robusto implicaría clases de excepción personalizadas y mensajes de error más detallados. * **Threading:** El servidor utiliza threads para manejar múltiples conexiones de clientes simultáneamente. Para aplicaciones de alto rendimiento, considera el uso de frameworks asíncronos como `asyncio`. * **Formato MCP:** El formato MCP basado en JSON es simple, pero se puede extender para incluir el control de versiones, metadatos y estructuras de datos más complejas. Protocol Buffers es una opción popular para la serialización eficiente y la definición de esquemas. * **Fuente de datos:** El `WeatherService` actualmente utiliza una fuente de datos simulada. En una aplicación real, te integrarías con una API meteorológica (por ejemplo, OpenWeatherMap, AccuWeather) o una base de datos. * **Logging:** El código utiliza el módulo `logging` para el registro básico. Configura los niveles y formatos de registro de forma adecuada para tus necesidades. * **Pruebas:** Los ejemplos de `pytest` demuestran las pruebas unitarias del `WeatherService` y el `MCPRequestHandler`. Agrega más pruebas para cubrir todos los aspectos de la funcionalidad del servidor. Considera las pruebas de integración para verificar la interacción con fuentes de datos externas. * **Pre-commit:** La configuración de pre-commit incluye hooks comunes para el formateo de código (Black), el linting (Flake8) y las comprobaciones básicas. Personaliza los hooks para que coincidan con los requisitos de tu proyecto. * **Configuración:** Utiliza variables de entorno o archivos de configuración para administrar configuraciones como host, puerto y claves de API. * **Seguridad:** Para entornos de producción, implementa medidas de seguridad como autenticación, autorización y validación de entrada para protegerte contra ataques maliciosos. * **ID de solicitud:** El `request_id` es crucial para correlacionar solicitudes y respuestas, especialmente en sistemas asíncronos o distribuidos. Permite al cliente hacer coincidir una respuesta con la solicitud original. El servidor siempre debe incluir el `request_id` en la respuesta. * **Códigos de error:** El uso de códigos de error estandarizados (como los de JSON-RPC) facilita a los clientes la comprensión de la naturaleza del error y su manejo adecuado. Se pueden utilizar códigos de error personalizados para errores específicos de la aplicación. * **Apagado elegante:** Implementa un mecanismo para apagar el servidor de forma elegante, permitiéndole terminar de procesar las solicitudes existentes antes de salir. Esto se puede hacer usando señales (por ejemplo, `SIGTERM`). **Cómo ejecutar** 1. Guarda los fragmentos de código como archivos Python separados (por ejemplo, `weather_server.py`, `weather_client.py`, `test_weather.py`). 2. Instala los paquetes requeridos: `pip install pytest pre-commit black flake8` 3. Ejecuta el servidor: `python weather_server.py` 4. Ejecuta el cliente en una terminal separada: `python weather_client.py` 5. Ejecuta las pruebas: `pytest` 6. Inicializa pre-commit: `pre-commit install` Este ejemplo completo proporciona una base sólida para construir un servidor MCP para datos meteorológicos. Recuerda adaptarlo y ampliarlo para satisfacer los requisitos específicos de tu aplicación.

Playcanvas_editor Mcp Server

Espejo de