HiMCP

Mcp_proxy_pydantic_agent

「MCPサーバーをPydanticエージェントに公開する例」

Tiny TODO MCP

AIアシスタントに永続的なタスク管理機能を提供し、通常は制限されるコンテキストを超えてタスクの作成、更新、追跡を可能にする、モデルコンテキストプロトコルサーバー。

Role-Specific Context MCP Server

AIエージェント向けのロールベースのコンテキスト管理を可能にするモデルコンテキストプロトコルサーバー。ユーザーは特定の指示を設定したり、分割されたメモリを維持したり、システム内のさまざまなエージェントロールに合わせてトーンを調整したりできます。

Vidu MCP Server

ViduのAIモデルを使用して、静止画から動画を生成できるサーバー。画像から動画への変換、タスクの監視、画像のアップロード機能などを搭載。

Medium MCP API Server

MediumのAPI用の、コンテンツ公開とユーザーアカウント管理を行うためのマイクロサービス通信プロトコル（MCP）サーバー

Linear

AIアシスタントがLinearプロジェクト管理システムと連携し、ユーザーが自然言語を通じて課題、プロジェクト、チームの取得、作成、更新を行えるようにする、モデルコンテキストプロトコルサーバー。

MCP Client Example ☀️

Okay, here's a basic example of a Python MCP (Memcached Client Protocol) client and server. Keep in mind that this is a *very* simplified example for demonstration purposes. A real-world Memcached server is significantly more complex. Also, for production use, you should use a well-established Memcached library like `pymemcache` or `python-memcached`. **Important Considerations:** * **Security:** This example is *not* secure. It doesn't handle authentication or encryption. Do not use it in a production environment without adding proper security measures. * **Error Handling:** The error handling is minimal. A real implementation would need much more robust error handling. * **Concurrency:** The server uses basic threading. For high-performance, consider using asynchronous I/O (e.g., `asyncio`). * **Protocol Completeness:** This example only implements the `set` and `get` commands. Memcached has many more commands. * **Data Serialization:** This example stores data as strings. You'll likely want to use a serialization library like `pickle` or `json` to store more complex data types. **1. Server (mcp_server.py):** ```python import socket import threading class MCPRequestHandler(threading.Thread): def __init__(self, client_socket, client_address, cache): threading.Thread.__init__(self) self.client_socket = client_socket self.client_address = client_address self.cache = cache print(f"Connection from {client_address}") def run(self): try: while True: data = self.client_socket.recv(1024).decode('utf-8').strip() if not data: break parts = data.split() command = parts[0].lower() if command == 'set': if len(parts) < 3: self.client_socket.sendall(b"ERROR: Invalid SET command\r\n") continue key = parts[1] value = " ".join(parts[2:]) # Handle values with spaces self.cache[key] = value self.client_socket.sendall(b"STORED\r\n") elif command == 'get': if len(parts) != 2: self.client_socket.sendall(b"ERROR: Invalid GET command\r\n") continue key = parts[1] if key in self.cache: value = self.cache[key] self.client_socket.sendall(f"VALUE {key} {len(value)}\r\n{value}\r\nEND\r\n".encode('utf-8')) else: self.client_socket.sendall(b"END\r\n") else: self.client_socket.sendall(b"ERROR: Unknown command\r\n") except Exception as e: print(f"Error handling client: {e}") finally: print(f"Closing connection from {self.client_address}") self.client_socket.close() class MCPServer: def __init__(self, host, port): self.host = host self.port = port self.cache = {} # In-memory cache 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) # Allow reuse of address self.server_socket.bind((self.host, self.port)) self.server_socket.listen(5) # Listen for up to 5 connections print(f"MCP Server listening on {self.host}:{self.port}") try: while True: client_socket, client_address = self.server_socket.accept() handler = MCPRequestHandler(client_socket, client_address, self.cache) handler.start() except KeyboardInterrupt: print("Server shutting down...") finally: if self.server_socket: self.server_socket.close() if __name__ == "__main__": HOST = "127.0.0.1" # Standard loopback interface address (localhost) PORT = 11211 # Standard Memcached port server = MCPServer(HOST, PORT) server.start() ``` **2. Client (mcp_client.py):** ```python import socket class MCPClient: def __init__(self, host, port): self.host = host self.port = port self.socket = None def connect(self): self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) try: self.socket.connect((self.host, self.port)) return True except socket.error as e: print(f"Connection error: {e}") self.socket = None return False def set(self, key, value): if not self.socket: if not self.connect(): return False command = f"set {key} {value}\r\n" try: self.socket.sendall(command.encode('utf-8')) response = self.socket.recv(1024).decode('utf-8').strip() return response == "STORED" except socket.error as e: print(f"Set error: {e}") return False def get(self, key): if not self.socket: if not self.connect(): return None command = f"get {key}\r\n" try: self.socket.sendall(command.encode('utf-8')) response = self.socket.recv(1024).decode('utf-8').strip() if response.startswith("VALUE"): parts = response.split() # key = parts[1] # Unused in this example, but available length = int(parts[2]) value = self.socket.recv(length).decode('utf-8') self.socket.recv(2) # Consume the "\r\n" after the value end_response = self.socket.recv(5).decode('utf-8').strip() # Expect "END" if end_response == "END": return value else: print(f"Unexpected response after value: {end_response}") return None elif response == "END": return None else: print(f"Unexpected response: {response}") return None except socket.error as e: print(f"Get error: {e}") return None def close(self): if self.socket: self.socket.close() self.socket = None if __name__ == "__main__": HOST = "127.0.0.1" PORT = 11211 client = MCPClient(HOST, PORT) if client.connect(): print("Connected to server.") # Set a value if client.set("mykey", "Hello, Memcached!"): print("Set 'mykey' successfully.") else: print("Failed to set 'mykey'.") # Get the value value = client.get("mykey") if value: print(f"Value for 'mykey': {value}") else: print("Key 'mykey' not found.") # Get a non-existent key value = client.get("nonexistent_key") if value: print(f"Value for 'nonexistent_key': {value}") else: print("Key 'nonexistent_key' not found.") client.close() print("Connection closed.") else: print("Failed to connect to server.") ``` **How to Run:** 1. **Save:** Save the code as `mcp_server.py` and `mcp_client.py`. 2. **Run the Server:** Open a terminal and run `python mcp_server.py`. The server will start listening on `127.0.0.1:11211`. 3. **Run the Client:** Open another terminal and run `python mcp_client.py`. The client will connect to the server, set a key, get the key, and then close the connection. **Explanation:** * **Server:** * `MCPServer` class: Handles the server socket, listening for connections, and creating a new thread for each client. * `MCPRequestHandler` class: Handles the communication with a single client. It receives data, parses the command (currently only `set` and `get`), and interacts with the in-memory `cache` (a Python dictionary). * The server uses a simple in-memory dictionary (`self.cache`) to store the data. * The server sends back responses like "STORED", "VALUE", and "END" according to the basic Memcached protocol. * **Client:** * `MCPClient` class: Handles the connection to the server and provides `set` and `get` methods. * The `set` method sends a "set" command to the server. * The `get` method sends a "get" command and parses the response from the server. * The client handles the basic protocol format, including the "VALUE", length, and "END" markers. **To Translate to Japanese:** Here's a translation of the comments and print statements in the code to Japanese. I'll provide the translated code snippets. You can replace the English text in the original code with these translations. **mcp_server.py (Japanese):** ```python import socket import threading class MCPRequestHandler(threading.Thread): def __init__(self, client_socket, client_address, cache): threading.Thread.__init__(self) self.client_socket = client_socket self.client_address = client_address self.cache = cache print(f"{client_address} からの接続") # Connection from {client_address} def run(self): try: while True: data = self.client_socket.recv(1024).decode('utf-8').strip() if not data: break parts = data.split() command = parts[0].lower() if command == 'set': if len(parts) < 3: self.client_socket.sendall(b"ERROR: 無効なSETコマンド\r\n") # ERROR: Invalid SET command continue key = parts[1] value = " ".join(parts[2:]) # Handle values with spaces self.cache[key] = value self.client_socket.sendall(b"STORED\r\n") elif command == 'get': if len(parts) != 2: self.client_socket.sendall(b"ERROR: 無効なGETコマンド\r\n") # ERROR: Invalid GET command continue key = parts[1] if key in self.cache: value = self.cache[key] self.client_socket.sendall(f"VALUE {key} {len(value)}\r\n{value}\r\nEND\r\n".encode('utf-8')) else: self.client_socket.sendall(b"END\r\n") else: self.client_socket.sendall(b"ERROR: 不明なコマンド\r\n") # ERROR: Unknown command except Exception as e: print(f"クライアント処理中にエラーが発生しました: {e}") # Error handling client: {e} finally: print(f"{self.client_address} からの接続を閉じます") # Closing connection from {self.client_address} self.client_socket.close() class MCPServer: def __init__(self, host, port): self.host = host self.port = port self.cache = {} # In-memory cache 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) # Allow reuse of address self.server_socket.bind((self.host, self.port)) self.server_socket.listen(5) # Listen for up to 5 connections print(f"MCPサーバーは {self.host}:{self.port} でリッスンしています") # MCP Server listening on {self.host}:{self.port} try: while True: client_socket, client_address = self.server_socket.accept() handler = MCPRequestHandler(client_socket, client_address, self.cache) handler.start() except KeyboardInterrupt: print("サーバーをシャットダウンしています...") # Server shutting down... finally: if self.server_socket: self.server_socket.close() if __name__ == "__main__": HOST = "127.0.0.1" # Standard loopback interface address (localhost) PORT = 11211 # Standard Memcached port server = MCPServer(HOST, PORT) server.start() ``` **mcp_client.py (Japanese):** ```python import socket class MCPClient: def __init__(self, host, port): self.host = host self.port = port self.socket = None def connect(self): self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) try: self.socket.connect((self.host, self.port)) return True except socket.error as e: print(f"接続エラー: {e}") # Connection error: {e} self.socket = None return False def set(self, key, value): if not self.socket: if not self.connect(): return False command = f"set {key} {value}\r\n" try: self.socket.sendall(command.encode('utf-8')) response = self.socket.recv(1024).decode('utf-8').strip() return response == "STORED" except socket.error as e: print(f"Setエラー: {e}") # Set error: {e} return False def get(self, key): if not self.socket: if not self.connect(): return None command = f"get {key}\r\n" try: self.socket.sendall(command.encode('utf-8')) response = self.socket.recv(1024).decode('utf-8').strip() if response.startswith("VALUE"): parts = response.split() # key = parts[1] # Unused in this example, but available length = int(parts[2]) value = self.socket.recv(length).decode('utf-8') self.socket.recv(2) # Consume the "\r\n" after the value end_response = self.socket.recv(5).decode('utf-8').strip() # Expect "END" if end_response == "END": return value else: print(f"値の後に予期しない応答: {end_response}") # Unexpected response after value: {end_response} return None elif response == "END": return None else: print(f"予期しない応答: {response}") # Unexpected response: {response} return None except socket.error as e: print(f"Getエラー: {e}") # Get error: {e} return None def close(self): if self.socket: self.socket.close() self.socket = None if __name__ == "__main__": HOST = "127.0.0.1" PORT = 11211 client = MCPClient(HOST, PORT) if client.connect(): print("サーバーに接続しました。") # Connected to server. # Set a value if client.set("mykey", "Hello, Memcached!"): print("'mykey' を正常に設定しました。") # Set 'mykey' successfully. else: print("'mykey' の設定に失敗しました。") # Failed to set 'mykey'. # Get the value value = client.get("mykey") if value: print(f"'mykey' の値: {value}") # Value for 'mykey': {value} else: print("キー 'mykey' は見つかりませんでした。") # Key 'mykey' not found. # Get a non-existent key value = client.get("nonexistent_key") if value: print(f"'nonexistent_key' の値: {value}") # Value for 'nonexistent_key': {value} else: print("キー 'nonexistent_key' は見つかりませんでした。") # Key 'nonexistent_key' not found. client.close() print("接続を閉じました。") # Connection closed. else: print("サーバーへの接続に失敗しました。") # Failed to connect to server. ``` **How to Use the Translated Code:** 1. Replace the English comments and `print` statements in `mcp_server.py` with the corresponding Japanese translations from the `mcp_server.py (Japanese)` section above. 2. Replace the English comments and `print` statements in `mcp_client.py` with the corresponding Japanese translations from the `mcp_client.py (Japanese)` section above. 3. Save the modified files. 4. Run the server and client as described earlier. The output will now be in Japanese. **Important Notes about Japanese:** * **Encoding:** Make sure your editor saves the files in UTF-8 encoding to properly display Japanese characters. * **Terminal Support:** Your terminal needs to support UTF-8 encoding to display Japanese characters correctly. Most modern terminals do, but you might need to configure it. * **Variable Names:** I've only translated the comments and print statements. You *could* also translate variable names (e.g., `client_socket` to `クライアントソケット`), but this is generally not recommended unless you are working on a project specifically for Japanese speakers. Keeping variable names in English makes the code more accessible to a wider audience. This provides a basic, functional MCP client and server example in Python, along with the Japanese translations you requested. Remember to use a proper Memcached library for production applications.