HiMCP

MCP Actions Adapter

MCPサーバーをGPT Actions互換APIに変換するシンプルなアダプター

Playcanvas_editor Mcp Server

鏡 (Kagami)

mcp-tools-cli

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

Polkassembly MCP Server

Polkassembly API 用の MCP サーバー

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サーバー

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

Atlassian Jira MCP Server

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

mcp-server

鏡 (Kagami)

IACR Cryptology ePrint Archive MCP Server

鏡 (Kagami)

MCP Server Playground

鏡 (Kagami)

Servidor MCP do Supabase

Supabase の MCP サーバー、データクエリと挿入機能付き

🌱 mcp-origin

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

📸 Smart Photo Journal MCP Server

鏡 (Kagami)

MSSQL MCP Server

鏡 (Kagami)

langchain-mcp

LangChain のための Model Context Protocol ツールサポート

YouTube MCP Server

鏡 (Kagami)

MCP Community Contributions

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

VoiceStudio MCP Server

MCP server for kintone

鏡 (Kagami)

MCP Servers Modified

Grasshopper MCP サーバー

Rhinoceros/Grasshopper 連携のための Model Context Protocol (MCP) サーバー実装。AI モデルがパラメトリックデザインツールと相互作用できるようにする。

mcptime

簡単なMCPサーバーで、現在の時刻を返すもの

RapidAPI MCP Server

承知いたしました。以下に、RapidAPI Global Patent APIとの連携とSQLiteストレージを利用したMCPサーバーの実装について、日本語で説明します。 **MCP (Minecraft Protocol) サーバー実装：RapidAPI Global Patent API連携とSQLiteストレージ** この実装は、Minecraft Protocol (MCP) サーバーを構築し、RapidAPI Global Patent APIを利用して特許情報を取得し、SQLiteデータベースに保存するものです。 **概要** 1. **MCPサーバー:** Minecraftクライアントからの接続を受け付け、ゲームロジックを処理するサーバーです。 2. **RapidAPI Global Patent API連携:** 特許情報を取得するために、RapidAPIのGlobal Patent APIを利用します。 3. **SQLiteストレージ:** 取得した特許情報をSQLiteデータベースに保存します。 **実装のポイント** * **MCPサーバーの構築:** * 適切なMCPサーバーライブラリ（例：Glowstone、SpongeForge）を選択し、基本的なサーバー機能を実装します。 * プレイヤーのログイン、チャット、コマンド処理などの機能を実装します。 * **RapidAPI Global Patent API連携:** * RapidAPIアカウントを作成し、Global Patent APIのサブスクリプションを取得します。 * APIキーを取得し、サーバーコードに組み込みます。 * APIリクエストを送信し、特許情報をJSON形式で取得する関数を実装します。 * APIリクエストのレート制限に注意し、適切なエラーハンドリングを実装します。 * **SQLiteストレージ:** * SQLiteデータベースを作成し、特許情報を格納するテーブルを定義します。 * 取得した特許情報をデータベースに挿入する関数を実装します。 * 特許情報を検索、更新、削除する関数を実装します。 * **Minecraftコマンドの実装:** * Minecraftクライアントから特許情報を検索するためのコマンドを実装します（例：`/patent search <キーワード>`）。 * コマンド実行時に、RapidAPI Global Patent APIにリクエストを送信し、取得した特許情報をSQLiteデータベースに保存します。 * 検索結果をMinecraftクライアントに表示します。 * **エラーハンドリング:** * APIリクエストのエラー、データベース操作のエラーなど、様々なエラーを適切に処理します。 * エラーログを記録し、問題発生時のデバッグを容易にします。 * **セキュリティ:** * APIキーを安全に管理します。 * SQLインジェクション攻撃を防ぐために、パラメータ化されたクエリを使用します。 **技術スタックの例** * **プログラミング言語:** Java (MCPサーバーライブラリとの互換性のため) * **MCPサーバーライブラリ:** Glowstone、SpongeForge * **データベース:** SQLite * **HTTPクライアント:** Apache HttpClient、OkHttp * **JSONライブラリ:** Gson、Jackson **実装例 (Java)** ```java // RapidAPI Global Patent APIへのリクエスト送信 public String searchPatents(String keyword) { // APIキー、APIエンドポイントの設定 String apiKey = "YOUR_API_KEY"; String apiUrl = "https://rapidapi.com/api/global-patent-api/search?q=" + keyword; // HTTPクライアントの作成 HttpClient client = HttpClientBuilder.create().build(); HttpGet request = new HttpGet(apiUrl); // ヘッダーの設定 request.setHeader("X-RapidAPI-Key", apiKey); request.setHeader("X-RapidAPI-Host", "global-patent-api.p.rapidapi.com"); // APIリクエストの送信 HttpResponse response = client.execute(request); // レスポンスの処理 HttpEntity entity = response.getEntity(); if (entity != null) { return EntityUtils.toString(entity); } else { return null; } } // SQLiteデータベースへの保存 public void savePatent(Patent patent) { // データベース接続 Connection connection = DriverManager.getConnection("jdbc:sqlite:patents.db"); // SQL文の作成 String sql = "INSERT INTO patents (title, abstract, publication_date) VALUES (?, ?, ?)"; PreparedStatement statement = connection.prepareStatement(sql); statement.setString(1, patent.getTitle()); statement.setString(2, patent.getAbstract()); statement.setString(3, patent.getPublicationDate()); // SQL文の実行 statement.executeUpdate(); // データベース接続のクローズ connection.close(); } ``` **注意点** * RapidAPI Global Patent APIの利用規約と料金プランをよく確認してください。 * APIのレート制限を超えないように、リクエスト頻度を調整してください。 * セキュリティ対策を徹底し、APIキーを安全に管理してください。この説明が、RapidAPI Global Patent APIとの連携とSQLiteストレージを利用したMCPサーバーの実装の参考になれば幸いです。具体的な実装方法については、各ライブラリのドキュメントやサンプルコードを参照してください。