MCP 服务器

发现优秀的 MCP 服务器

通过 MCP 服务器扩展您的代理能力，拥有 12,252 个能力。

研究与数据1,246

서울시 교통 데이터 MCP 서버

首尔市交通数据 MCP 服务器 - 提供实时交通信息、公共交通、Ttareungi（首尔市公共自行车）等数据的 MCP 服务器

Weather App

Okay, this is a complex request, but I'll provide a conceptual outline and code snippets to illustrate how you might implement a minimal MCP (Microservice Communication Protocol) server for weather data in Python, along with testing and pre-commit setup. I'll focus on clarity and structure rather than production-readiness. **Conceptual Overview** 1. **MCP Definition (Protocol Buffer):** Define the data structures and service interface using Protocol Buffers. This is the core of MCP. 2. **Server Implementation (Python):** Implement a gRPC server in Python that serves weather data based on the MCP definition. 3. **Client (Example):** A simple client to demonstrate how to request data from the server. 4. **Testing (pytest):** Write unit tests to verify the server's functionality. 5. **Pre-commit Hooks:** Configure pre-commit to run linters (e.g., `flake8`, `black`) and tests before committing code. **1. MCP Definition (weather.proto)** Create a file named `weather.proto`: ```protobuf syntax = "proto3"; package weather; service WeatherService { rpc GetWeather(WeatherRequest) returns (WeatherResponse) {} } message WeatherRequest { string city = 1; } message WeatherResponse { string city = 1; float temperature = 2; string condition = 3; } ``` **Explanation:** * `syntax = "proto3";`: Specifies the Protocol Buffer version. * `package weather;`: Defines the package name for the generated code. * `service WeatherService`: Defines the service with a single RPC method `GetWeather`. * `WeatherRequest`: The request message, containing the city name. * `WeatherResponse`: The response message, containing weather information. **2. Generate gRPC Code (Python)** You'll need the `grpcio-tools` package: ```bash pip install grpcio-tools ``` Then, compile the `.proto` file: ```bash python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. weather.proto ``` This will generate two Python files: `weather_pb2.py` (the data structures) and `weather_pb2_grpc.py` (the gRPC service definitions). **3. Server Implementation (server.py)** ```python import grpc from concurrent import futures import weather_pb2 import weather_pb2_grpc class WeatherServicer(weather_pb2_grpc.WeatherServiceServicer): def GetWeather(self, request, context): city = request.city # Simulate weather data (replace with actual data source) if city == "London": temperature = 15.0 condition = "Cloudy" elif city == "New York": temperature = 25.0 condition = "Sunny" else: temperature = 10.0 condition = "Unknown" return weather_pb2.WeatherResponse(city=city, temperature=temperature, condition=condition) def serve(): server = grpc.server(futures.ThreadPoolExecutor(max_workers=10)) weather_pb2_grpc.add_WeatherServiceServicer_to_server(WeatherServicer(), server) server.add_insecure_port('[::]:50051') # Listen on all interfaces, port 50051 server.start() server.wait_for_termination() if __name__ == '__main__': print("Starting gRPC server...") serve() ``` **Explanation:** * `WeatherServicer`: Implements the `WeatherServiceServicer` interface defined in `weather_pb2_grpc.py`. * `GetWeather`: The actual implementation of the RPC method. It receives a `WeatherRequest` and returns a `WeatherResponse`. This example uses hardcoded data for simplicity. In a real application, you'd fetch weather data from an external API or database. * `serve`: Creates a gRPC server, registers the `WeatherServicer`, and starts the server. **4. Client (client.py)** ```python import grpc import weather_pb2 import weather_pb2_grpc def run(): with grpc.insecure_channel('localhost:50051') as channel: stub = weather_pb2_grpc.WeatherServiceStub(channel) city = "London" request = weather_pb2.WeatherRequest(city=city) response = stub.GetWeather(request) print(f"Weather in {response.city}:") print(f" Temperature: {response.temperature}°C") print(f" Condition: {response.condition}") if __name__ == '__main__': run() ``` **Explanation:** * Creates a gRPC channel to connect to the server. * Creates a `WeatherServiceStub` to make RPC calls. * Sends a `WeatherRequest` for the city "London". * Prints the received `WeatherResponse`. **5. Testing (test_server.py)** Install `pytest` and `grpcio-testing`: ```bash pip install pytest grpcio-testing ``` ```python import pytest import grpc from grpc import aio from concurrent import futures import weather_pb2 import weather_pb2_grpc from server import WeatherServicer # Import your server implementation @pytest.fixture def grpc_server(): server = grpc.server(futures.ThreadPoolExecutor(max_workers=10)) weather_pb2_grpc.add_WeatherServiceServicer_to_server(WeatherServicer(), server) port = server.add_insecure_port('[::]:0') # Let the OS choose a port server.start() yield server, port # Provide the server and port to the tests server.stop(0) def test_get_weather(grpc_server): server, port = grpc_server with grpc.insecure_channel(f'localhost:{port}') as channel: stub = weather_pb2_grpc.WeatherServiceStub(channel) request = weather_pb2.WeatherRequest(city="London") response = stub.GetWeather(request) assert response.city == "London" assert response.temperature == 15.0 assert response.condition == "Cloudy" def test_get_weather_new_york(grpc_server): server, port = grpc_server with grpc.insecure_channel(f'localhost:{port}') as channel: stub = weather_pb2_grpc.WeatherServiceStub(channel) request = weather_pb2.WeatherRequest(city="New York") response = stub.GetWeather(request) assert response.city == "New York" assert response.temperature == 25.0 assert response.condition == "Sunny" def test_get_weather_unknown(grpc_server): server, port = grpc_server with grpc.insecure_channel(f'localhost:{port}') as channel: stub = weather_pb2_grpc.WeatherServiceStub(channel) request = weather_pb2.WeatherRequest(city="SomeRandomCity") response = stub.GetWeather(request) assert response.city == "SomeRandomCity" assert response.temperature == 10.0 assert response.condition == "Unknown" ``` **Explanation:** * `grpc_server` fixture: Creates and starts a gRPC server before each test and stops it afterward. It uses a dynamically assigned port to avoid conflicts. * `test_get_weather`: Tests the `GetWeather` method for a specific city ("London"). * `test_get_weather_new_york`: Tests the `GetWeather` method for a specific city ("New York"). * `test_get_weather_unknown`: Tests the `GetWeather` method for an unknown city. Run the tests: ```bash pytest ``` **6. Pre-commit Setup** Install `pre-commit`: ```bash pip install pre-commit ``` Create a `.pre-commit-config.yaml` file in the root of your project: ```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 - 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 additional_dependencies: ["flake8-protobuf"] # Add protobuf support - repo: local hooks: - id: pytest name: pytest entry: pytest language: system types: [python] pass_filenames: false ``` **Explanation:** * This configuration uses several popular pre-commit hooks: * `trailing-whitespace`, `end-of-file-fixer`, `check-yaml`, `check-added-large-files`: Basic cleanup and checks. * `black`: Code formatter. * `flake8`: Linter. The `additional_dependencies` line adds support for linting Protocol Buffer-related code. * `pytest`: Runs your tests. Install the pre-commit hooks: ```bash pre-commit install ``` Now, every time you commit, pre-commit will run these checks and formatters. If any of them fail, the commit will be aborted. **Running the Example** 1. **Generate gRPC code:** `python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. weather.proto` 2. **Start the server:** `python server.py` (in a separate terminal) 3. **Run the client:** `python client.py` 4. **Run the tests:** `pytest` 5. **Commit your changes:** `git commit -m "Your commit message"` (pre-commit will run) **Important Considerations and Improvements** * **Error Handling:** The server implementation lacks proper error handling. You should handle exceptions and return appropriate gRPC status codes. * **Data Source:** The server uses hardcoded weather data. Replace this with a real data source (e.g., an API or database). * **Authentication/Authorization:** For production systems, you'll need to add authentication and authorization to secure your gRPC service. * **Logging:** Implement logging for debugging and monitoring. * **Configuration:** Use environment variables or configuration files to manage server settings (e.g., port number, data source connection details). * **Asynchronous gRPC:** Consider using `grpc.aio` for asynchronous gRPC, which can improve performance, especially for I/O-bound operations. * **Health Checks:** Implement a health check endpoint to allow monitoring systems to verify the server's availability. * **Code Generation:** Consider using a build system (e.g., `make`, `nox`) to automate the code generation process. * **Dependency Management:** Use a tool like `poetry` or `pipenv` to manage your project's dependencies. * **CI/CD:** Integrate your tests and pre-commit hooks into a CI/CD pipeline to automate the build, test, and deployment process. This comprehensive example provides a solid foundation for building a gRPC-based microservice with testing and pre-commit setup. Remember to adapt and extend it to meet the specific requirements of your application.

📸 Smart Photo Journal MCP Server

镜子 (jìng zi)

RapidAPI MCP Server

以下是将 RapidAPI Global Patent API 集成到 MCP 服务器并使用 SQLite 存储的实现方案： **标题：基于 RapidAPI Global Patent API 的 MCP 服务器实现，并使用 SQLite 存储** **概述：** 本方案描述了如何构建一个 MCP (Minimal Control Protocol) 服务器，该服务器利用 RapidAPI 的 Global Patent API 来检索专利数据，并将检索到的数据存储在 SQLite 数据库中。该服务器将接收来自客户端的请求，调用 RapidAPI 的 Global Patent API，解析响应，并将结果存储到 SQLite 数据库中。 **组件：** * **MCP 服务器:** 负责监听客户端请求，处理请求，调用 RapidAPI，并将结果返回给客户端。 * **RapidAPI Global Patent API:** 提供专利数据检索服务。 * **SQLite 数据库:** 用于存储检索到的专利数据。 * **客户端:** 发送请求到 MCP 服务器，并接收响应。 **技术栈：** * **编程语言:** Python (推荐，易于使用，拥有丰富的库) * **MCP 框架:** 可以使用现有的 MCP 框架，或者自行实现简单的 MCP 协议处理。 * **RapidAPI 客户端库:** 可以使用 `requests` 库或其他 HTTP 客户端库来调用 RapidAPI。 * **SQLite 数据库库:** `sqlite3` (Python 内置) **实现步骤：** 1. **环境搭建:** * 安装 Python。 * 安装必要的 Python 库：`pip install requests` * 安装 SQLite (通常操作系统自带)。 2. **RapidAPI 密钥获取:** * 在 RapidAPI 注册并订阅 Global Patent API。 * 获取 API 密钥 (通常在 RapidAPI 控制面板中)。 3. **SQLite 数据库设计:** * 设计数据库表结构，用于存储专利数据。例如： * `patents` 表： * `id` (INTEGER PRIMARY KEY AUTOINCREMENT) * `patent_number` (TEXT UNIQUE) * `title` (TEXT) * `abstract` (TEXT) * `inventors` (TEXT) * `assignees` (TEXT) * `filing_date` (TEXT) * `publication_date` (TEXT) * `raw_data` (TEXT) (存储完整的 API 响应，方便后续处理) * `created_at` (TIMESTAMP DEFAULT CURRENT_TIMESTAMP) * 创建数据库和表。 4. **MCP 服务器实现:** * **监听端口:** 服务器监听指定的端口，等待客户端连接。 * **接收请求:** 接收客户端发送的 MCP 请求。 * **解析请求:** 解析 MCP 请求，提取请求参数 (例如，搜索关键词，专利号等)。 * **调用 RapidAPI:** 使用 RapidAPI 密钥和请求参数，调用 Global Patent API。 * **解析响应:** 解析 RapidAPI 返回的 JSON 响应。 * **存储数据到 SQLite:** 将解析后的专利数据存储到 SQLite 数据库中。 * **返回响应:** 将结果 (例如，成功/失败，专利 ID 等) 封装成 MCP 响应，返回给客户端。 5. **客户端实现 (可选):** * 编写客户端程序，用于发送 MCP 请求到服务器，并接收响应。 **代码示例 (Python):** ```python import sqlite3 import requests import json # RapidAPI 配置 RAPIDAPI_KEY = "YOUR_RAPIDAPI_KEY" # 替换为你的 RapidAPI 密钥 RAPIDAPI_HOST = "global-patent.p.rapidapi.com" RAPIDAPI_ENDPOINT = "/patents/search" # SQLite 数据库配置 DATABASE_FILE = "patents.db" def create_table(): conn = sqlite3.connect(DATABASE_FILE) cursor = conn.cursor() cursor.execute(""" CREATE TABLE IF NOT EXISTS patents ( id INTEGER PRIMARY KEY AUTOINCREMENT, patent_number TEXT UNIQUE, title TEXT, abstract TEXT, inventors TEXT, assignees TEXT, filing_date TEXT, publication_date TEXT, raw_data TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) conn.commit() conn.close() def search_patents(query): url = f"https://{RAPIDAPI_HOST}{RAPIDAPI_ENDPOINT}" headers = { "X-RapidAPI-Key": RAPIDAPI_KEY, "X-RapidAPI-Host": RAPIDAPI_HOST } querystring = {"q": query} response = requests.get(url, headers=headers, params=querystring) if response.status_code == 200: return response.json() else: print(f"Error: {response.status_code} - {response.text}") return None def store_patent_data(patent_data): conn = sqlite3.connect(DATABASE_FILE) cursor = conn.cursor() try: patent_number = patent_data.get("patent_number") title = patent_data.get("title") abstract = patent_data.get("abstract") inventors = ", ".join(patent_data.get("inventors", [])) assignees = ", ".join(patent_data.get("assignees", [])) filing_date = patent_data.get("filing_date") publication_date = patent_data.get("publication_date") raw_data = json.dumps(patent_data) cursor.execute(""" INSERT INTO patents (patent_number, title, abstract, inventors, assignees, filing_date, publication_date, raw_data) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, (patent_number, title, abstract, inventors, assignees, filing_date, publication_date, raw_data)) conn.commit() print(f"Patent {patent_number} stored successfully.") except sqlite3.IntegrityError: print(f"Patent {patent_number} already exists in the database.") except Exception as e: print(f"Error storing patent data: {e}") conn.rollback() finally: conn.close() # 示例 MCP 服务器 (简化版) def handle_mcp_request(request): # 假设请求格式为 "SEARCH:关键词" if request.startswith("SEARCH:"): query = request[7:] print(f"Searching for: {query}") patent_results = search_patents(query) if patent_results and patent_results.get("results"): for patent in patent_results["results"]: store_patent_data(patent) return "SEARCH_SUCCESS" else: return "SEARCH_FAILED" else: return "INVALID_REQUEST" # 主程序 if __name__ == "__main__": create_table() # 创建数据库表 # 模拟 MCP 请求 mcp_request = "SEARCH:artificial intelligence" response = handle_mcp_request(mcp_request) print(f"MCP Response: {response}") mcp_request = "SEARCH:blockchain technology" response = handle_mcp_request(mcp_request) print(f"MCP Response: {response}") print("Done.") ``` **关键点：** * **错误处理:** 在调用 RapidAPI 和操作 SQLite 数据库时，需要进行充分的错误处理，例如处理网络错误，API 错误，数据库错误等。 * **数据清洗:** RapidAPI 返回的数据可能需要进行清洗和转换，才能更好地存储到数据库中。 * **并发处理:** 如果服务器需要处理大量的并发请求，需要考虑使用多线程或异步编程来提高性能。 * **安全性:** 保护 RapidAPI 密钥，避免泄露。对客户端请求进行验证，防止恶意请求。 * **MCP 协议:** 根据实际需求，定义合适的 MCP 协议，包括请求格式，响应格式，错误码等。 * **分页处理:** 如果 RapidAPI 返回的结果集很大，需要进行分页处理，避免一次性加载所有数据。 * **速率限制:** 注意 RapidAPI 的速率限制，避免超过限制导致请求失败。可以实现速率限制机制，例如使用令牌桶算法。 * **数据更新:** 可以定期更新数据库中的专利数据，以保持数据的最新性。 **改进方向：** * **使用更完善的 MCP 框架:** 例如 Twisted, asyncio 等。 * **添加缓存机制:** 使用 Redis 或 Memcached 等缓存数据库，缓存常用的查询结果，提高响应速度。 * **实现更复杂的搜索功能:** 例如，支持按发明人，申请人，申请日期等进行搜索。 * **提供 API 接口:** 将 MCP 服务器封装成 RESTful API，方便其他应用程序调用。 * **使用 ORM:** 使用 SQLAlchemy 等 ORM 框架，简化数据库操作。 **总结：** 本方案提供了一个将 RapidAPI Global Patent API 集成到 MCP 服务器并使用 SQLite 存储的基本框架。您可以根据实际需求，对该方案进行修改和扩展，以满足您的特定需求。请务必替换 `YOUR_RAPIDAPI_KEY` 为您自己的 RapidAPI 密钥。这个示例代码只是一个起点，你需要根据你的具体需求进行修改和完善。