HiMCP

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

D-Zeroのフロントエンドコーディング用MCPサーバー

Loxo MCP Server

鏡 (Kagami)

Sequential Thinking MCP Server

Polkassembly MCP Server

Polkassembly API 用の MCP サーバー

MCP Community Contributions

これは、MCPに関連するあらゆるもの（サーバー、クライアント、およびMCP周辺のプロジェクト）のディレクトリリポジトリです。

VoiceStudio MCP Server

MCP server for kintone

鏡 (Kagami)

Weather App

Okay, here's a conceptual outline and example code snippets for a minimal MCP (Microservice Communication Protocol) server implementation in Python for weather data, along with testing and pre-commit setup. This is a simplified example to illustrate the core concepts. A production-ready system would require more robust error handling, security, and scalability considerations. **Conceptual Outline** 1. **MCP Server (Weather Service):** * Listens for MCP requests on a specific port. * Receives requests for weather data (e.g., by location). * Retrieves weather data (from a mock source or a real weather API). * Encodes the weather data into an MCP-compatible response. * Sends the response back to the client. 2. **MCP Protocol:** * We'll define a simple MCP protocol for this example. It will use JSON for data serialization. * **Request Format:** ```json { "method": "get_weather", "params": { "location": "London" } } ``` * **Response Format (Success):** ```json { "result": { "temperature": 15, "condition": "Cloudy" }, "error": null } ``` * **Response Format (Error):** ```json { "result": null, "error": { "code": -1, "message": "Location not found" } } ``` 3. **Testing:** * Unit tests to verify the server's behavior (e.g., correct responses for valid locations, error handling for invalid locations). 4. **Pre-commit Hooks:** * Automated checks (e.g., code formatting, linting, running tests) before committing code to the repository. **Example Code (Python)** ```python import socket import json import threading import time import random # For mock weather data import logging import os # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') HOST = '127.0.0.1' # Standard loopback interface address (localhost) PORT = 65432 # Port to listen on (non-privileged ports are > 1023) # Mock Weather Data (Replace with a real API call in a real application) WEATHER_DATA = { "London": {"temperature": 12, "condition": "Rainy"}, "Tokyo": {"temperature": 25, "condition": "Sunny"}, "New York": {"temperature": 18, "condition": "Partly Cloudy"}, "Paris": {"temperature": 15, "condition": "Cloudy"}, } def get_weather(location): """Retrieves weather data for a given location.""" if location in WEATHER_DATA: return WEATHER_DATA[location] else: return None def handle_client(conn, addr): """Handles communication with a single client.""" logging.info(f"Connected by {addr}") try: while True: data = conn.recv(1024) # Receive data in 1024-byte chunks if not data: break try: request = json.loads(data.decode('utf-8')) logging.info(f"Received request: {request}") if request.get("method") == "get_weather": location = request.get("params", {}).get("location") if location: weather_data = get_weather(location) if weather_data: response = {"result": weather_data, "error": None} else: response = {"result": None, "error": {"code": -1, "message": "Location not found"}} else: response = {"result": None, "error": {"code": -2, "message": "Location parameter missing"}} else: response = {"result": None, "error": {"code": -3, "message": "Invalid method"}} conn.sendall(json.dumps(response).encode('utf-8')) logging.info(f"Sent response: {response}") except json.JSONDecodeError: logging.error("Invalid JSON received") response = {"result": None, "error": {"code": -4, "message": "Invalid JSON"}} conn.sendall(json.dumps(response).encode('utf-8')) except Exception as e: logging.exception("An error occurred while processing the request") response = {"result": None, "error": {"code": -5, "message": f"Server error: {str(e)}"}} conn.sendall(json.dumps(response).encode('utf-8')) except ConnectionResetError: logging.warning(f"Connection reset by {addr}") except Exception as e: logging.exception(f"Error handling client {addr}: {e}") finally: conn.close() logging.info(f"Connection closed with {addr}") def start_server(): """Starts the MCP server.""" try: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) # Allow reuse of the address s.bind((HOST, PORT)) s.listen() logging.info(f"Server listening on {HOST}:{PORT}") while True: conn, addr = s.accept() thread = threading.Thread(target=handle_client, args=(conn, addr)) thread.start() except Exception as e: logging.error(f"Server error: {e}") finally: logging.info("Server shutting down.") if __name__ == "__main__": start_server() ``` **Example Client (Python)** ```python import socket import json HOST = '127.0.0.1' # The server's hostname or IP address PORT = 65432 # The port used by the server def get_weather_from_server(location): """Sends a request to the server and retrieves weather data.""" try: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.connect((HOST, PORT)) request = { "method": "get_weather", "params": { "location": location } } s.sendall(json.dumps(request).encode('utf-8')) data = s.recv(1024) response = json.loads(data.decode('utf-8')) if response.get("error"): print(f"Error: {response['error']['message']}") return None else: return response["result"] except ConnectionRefusedError: print("Error: Could not connect to the server. Is it running?") return None except Exception as e: print(f"Error: {e}") return None if __name__ == "__main__": location = "London" weather = get_weather_from_server(location) if weather: print(f"Weather in {location}: {weather}") location = "Atlantis" weather = get_weather_from_server(location) if weather: print(f"Weather in {location}: {weather}") ``` **Testing (pytest)** ```python import pytest import socket import json import threading import time from your_server_file import start_server # Replace your_server_file HOST = '127.0.0.1' PORT = 65433 # Use a different port for testing to avoid conflicts @pytest.fixture(scope="module") def server(): """Starts the server in a separate thread and shuts it down after tests.""" server_thread = threading.Thread(target=start_server) server_thread.daemon = True # Allow the main thread to exit even if the server thread is running server_thread.start() time.sleep(0.1) # Give the server a moment to start yield # Add shutdown logic here if needed. For this simple example, it's not strictly necessary # because the server thread is a daemon thread. In a more complex server, you'd want # a proper shutdown mechanism. def send_request(request): """Sends a request to the server and returns the response.""" try: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.connect((HOST, PORT)) s.sendall(json.dumps(request).encode('utf-8')) data = s.recv(1024) return json.loads(data.decode('utf-8')) except Exception as e: pytest.fail(f"Error communicating with server: {e}") def test_get_weather_valid_location(server): """Tests getting weather for a valid location.""" request = { "method": "get_weather", "params": { "location": "London" } } response = send_request(request) assert response["error"] is None assert response["result"]["temperature"] == 12 assert response["result"]["condition"] == "Rainy" def test_get_weather_invalid_location(server): """Tests getting weather for an invalid location.""" request = { "method": "get_weather", "params": { "location": "Atlantis" } } response = send_request(request) assert response["result"] is None assert response["error"]["code"] == -1 assert response["error"]["message"] == "Location not found" def test_get_weather_missing_location(server): """Tests getting weather with a missing location parameter.""" request = { "method": "get_weather", "params": {} } response = send_request(request) assert response["result"] is None assert response["error"]["code"] == -2 assert response["error"]["message"] == "Location parameter missing" def test_invalid_method(server): """Tests calling an invalid method.""" request = { "method": "get_forecast", "params": { "location": "London" } } response = send_request(request) assert response["result"] is None assert response["error"]["code"] == -3 assert response["error"]["message"] == "Invalid method" def test_invalid_json(server): """Tests sending invalid JSON to the server.""" try: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.connect((HOST, PORT)) s.sendall(b"This is not JSON") data = s.recv(1024) response = json.loads(data.decode('utf-8')) assert response["result"] is None assert response["error"]["code"] == -4 assert response["error"]["message"] == "Invalid JSON" except Exception as e: pytest.fail(f"Error communicating with server: {e}") ``` **To run the tests:** 1. Save the server code as `your_server_file.py` (or whatever you want to name it). 2. Save the test code as `test_your_server_file.py`. 3. Install pytest: `pip install pytest` 4. Run the tests from the command line: `pytest` **Pre-commit Setup** 1. **Install pre-commit:** ```bash pip install pre-commit ``` 2. **Create a `.pre-commit-config.yaml` file in the root of your repository:** ```yaml repos: - repo: https://github.com/pre-commit/pre-commit-hooks rev: v4.5.0 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 hooks: - id: black - repo: https://github.com/PyCQA/flake8 rev: 7.0.0 hooks: - id: flake8 args: ["--max-line-length=120"] # Adjust as needed - repo: https://github.com/pre-commit/mirrors-mypy rev: v1.8.0 hooks: - id: mypy additional_dependencies: - types-requests - pytest exclude: ^tests/ # Exclude tests directory from mypy checks ``` 3. **Install the pre-commit hooks:** ```bash pre-commit install ``` Now, every time you try to commit code, pre-commit will run the configured checks (e.g., formatting with Black, linting with Flake8, type checking with MyPy, running tests). If any of the checks fail, the commit will be aborted, and you'll need to fix the issues before committing again. **Explanation and Improvements** * **Error Handling:** The server includes basic error handling for invalid requests, missing parameters, and server-side exceptions. More robust error handling (e.g., logging detailed error information, using custom exception classes) would be needed for a production system. * **Threading:** The server uses threads to handle multiple client connections concurrently. For very high concurrency, consider using an asynchronous framework like `asyncio`. * **MCP Protocol:** The example uses a simple JSON-based MCP protocol. For more complex scenarios, you might consider using a more efficient binary serialization format (e.g., Protocol Buffers, MessagePack) or a more formal RPC framework (e.g., gRPC). * **Weather Data Source:** The example uses mock weather data. In a real application, you would integrate with a weather API (e.g., OpenWeatherMap, AccuWeather). Consider using environment variables to store API keys. * **Testing:** The tests cover basic scenarios. Add more tests to cover edge cases, error conditions, and different input values. Consider using mocking to isolate the server from external dependencies (e.g., the weather API). * **Logging:** The server uses the `logging` module to log events. Configure logging appropriately for your environment (e.g., log to a file, use different log levels). * **Configuration:** Use environment variables or a configuration file to store settings like the host, port, and API keys. * **Security:** For a production system, implement appropriate security measures (e.g., authentication, authorization, encryption). * **Deployment:** Consider using a containerization technology like Docker to package and deploy the server. * **Type Hinting:** The example uses type hints. This helps to improve code readability and maintainability. Use a type checker like MyPy to verify type correctness. * **Pre-commit:** The pre-commit configuration includes several useful hooks. Customize the configuration to suit your project's needs. Consider adding hooks for running tests and checking for security vulnerabilities. * **Shutdown:** The server example doesn't have a graceful shutdown mechanism. In a production environment, you'd want to handle signals (e.g., SIGTERM, SIGINT) to shut down the server cleanly. This might involve closing sockets, releasing resources, and finishing any in-progress requests. This comprehensive example provides a solid foundation for building an MCP server for weather data. Remember to adapt and extend it based on your specific requirements.

MCP Servers Modified

Playcanvas_editor Mcp Server

鏡 (Kagami)

mcp-tools-cli

Model Context Protocol (MCP) サーバーとやり取りするためのコマンドラインクライアント

MCP Server Playground

鏡 (Kagami)

AB Component Server

🌱 mcp-origin

MCPサーバーを管理するMCPサーバー

Atlassian Jira MCP Server

Atlassian Jira向けのNode.js/TypeScript製MCPサーバー。AIシステム（LLM）に、プロジェクトのリスト/取得、課題の検索/取得（JQL/IDを使用）、および開発情報（コミット、PR）の表示といったツールを提供します。AI機能をJiraのプロジェクト管理および課題追跡ワークフローに直接接続します。

mcp-server

鏡 (Kagami)