Discover Awesome MCP Servers

📸 Smart Photo Journal MCP Server

Cermin dari

Weather App

Okay, here's an example of a minimal MCP (Microservice Communication Protocol) server implementation in Python for weather data, along with basic testing and pre-commit setup. This is a simplified example to illustrate the core concepts. You'll likely need to adapt it to your specific needs. **1. Project Structure:** ``` weather_service/ ├── pyproject.toml ├── src/ │ └── weather_service/ │ ├── __init__.py │ ├── main.py │ ├── weather.py │ └── models.py ├── tests/ │ └── test_weather.py ├── .pre-commit-config.yaml └── README.md ``` **2. `pyproject.toml` (Poetry configuration):** ```toml [tool.poetry] name = "weather-service" version = "0.1.0" description = "A simple weather service using MCP" authors = ["Your Name <your.email@example.com>"] readme = "README.md" [tool.poetry.dependencies] python = "^3.9" fastapi = "^0.100.0" uvicorn = "^0.23.0" pydantic = "^2.0" mcp-lib = "^0.1.0" # Replace with your actual MCP library [tool.poetry.dev-dependencies] pytest = "^7.0" pytest-cov = "^4.0" pre-commit = "^3.0" flake8 = "^6.0" mypy = "^1.0" [build-system] requires = ["poetry-core"] build-backend = "poetry.core.masonry.api" [tool.pytest.ini_options] testpaths = ["tests"] addopts = "--cov=src/weather_service --cov-report term-missing" [tool.mypy] strict = true ignore_missing_imports = true ``` **Explanation:** * **`[tool.poetry]`:** Defines project metadata. * **`[tool.poetry.dependencies]`:** Lists runtime dependencies: * `fastapi`: A modern, fast (high-performance), web framework for building APIs. * `uvicorn`: An ASGI (Asynchronous Server Gateway Interface) server for running FastAPI applications. * `pydantic`: Data validation and settings management using Python type annotations. * `mcp-lib`: **Crucially, replace this with the actual name of your MCP library.** This is the library that handles the MCP communication. I'm assuming you have a library that provides the necessary functions for sending and receiving MCP messages. * **`[tool.poetry.dev-dependencies]`:** Lists development dependencies: * `pytest`: A testing framework. * `pytest-cov`: A pytest plugin for coverage reporting. * `pre-commit`: A framework for managing pre-commit hooks. * `flake8`: A code style checker. * `mypy`: A static type checker. * **`[tool.pytest.ini_options]`:** Configures pytest. * **`[tool.mypy]`:** Configures mypy. **Install dependencies:** ```bash poetry install ``` **3. `src/weather_service/models.py` (Data Models):** ```python from pydantic import BaseModel class WeatherData(BaseModel): city: str temperature: float condition: str ``` **Explanation:** * Uses Pydantic to define a `WeatherData` model. This model will be used to structure the weather data that the service returns. **4. `src/weather_service/weather.py` (Weather Logic):** ```python from .models import WeatherData import random def get_weather(city: str) -> WeatherData: """ Simulates fetching weather data for a given city. """ temperature = random.uniform(10, 35) # Simulate temperature between 10 and 35 conditions = ["Sunny", "Cloudy", "Rainy", "Snowy"] condition = random.choice(conditions) return WeatherData(city=city, temperature=temperature, condition=condition) ``` **Explanation:** * `get_weather(city: str)`: This function simulates fetching weather data. In a real application, you would replace this with a call to an external weather API or database. It returns a `WeatherData` object. **5. `src/weather_service/main.py` (FastAPI Application and MCP Integration):** ```python from fastapi import FastAPI from .weather import get_weather from .models import WeatherData import asyncio import mcp_lib # Replace with your actual MCP library app = FastAPI() # MCP Configuration (Replace with your actual configuration) MCP_SERVER_ADDRESS = "localhost" MCP_SERVER_PORT = 5000 async def handle_weather_request(request_data: dict) -> dict: """ Handles incoming weather requests via MCP. """ city = request_data.get("city") if not city: return {"error": "City is required"} weather_data = get_weather(city) return weather_data.dict() # Convert to dictionary for MCP async def mcp_server(): """ Starts the MCP server and listens for weather requests. """ async def message_handler(message: dict) -> dict: """Handles incoming MCP messages.""" print(f"Received MCP message: {message}") return await handle_weather_request(message) await mcp_lib.start_server(MCP_SERVER_ADDRESS, MCP_SERVER_PORT, message_handler) # Replace with your MCP server start function @app.on_event("startup") async def startup_event(): """Starts the MCP server on application startup.""" asyncio.create_task(mcp_server()) @app.get("/weather/{city}") async def get_weather_http(city: str) -> WeatherData: """ HTTP endpoint for getting weather data. """ return get_weather(city) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000) ``` **Explanation:** * **FastAPI Setup:** * Creates a FastAPI application instance. * Defines an HTTP endpoint `/weather/{city}` that calls the `get_weather` function and returns the result. * **MCP Integration:** * `MCP_SERVER_ADDRESS` and `MCP_SERVER_PORT`: Configuration for the MCP server. **Replace these with your actual configuration.** * `handle_weather_request(request_data: dict)`: This function is the core of the MCP integration. It receives a dictionary representing the MCP request, extracts the `city` from the request, calls the `get_weather` function, and returns the weather data as a dictionary. **This is where you would adapt the code to handle the specific format of your MCP messages.** * `mcp_server()`: This function starts the MCP server. **You need to replace `mcp_lib.start_server()` with the actual function from your MCP library that starts the server.** The `message_handler` function is called whenever an MCP message is received. * `startup_event()`: This function is called when the FastAPI application starts. It starts the MCP server in a background task using `asyncio.create_task()`. * **Important:** This example assumes that your MCP library provides a function like `mcp_lib.start_server()` that takes an address, port, and a message handler function as arguments. You will need to adapt this code to match the API of your MCP library. The `message_handler` function should take the incoming MCP message as input and return the response. **6. `tests/test_weather.py` (Tests):** ```python from src.weather_service.weather import get_weather from src.weather_service.models import WeatherData def test_get_weather(): weather_data = get_weather("London") assert isinstance(weather_data, WeatherData) assert weather_data.city == "London" assert isinstance(weather_data.temperature, float) assert isinstance(weather_data.condition, str) ``` **Explanation:** * A simple test case that checks if the `get_weather` function returns a `WeatherData` object with the correct city and data types. **Run tests:** ```bash poetry run pytest ``` **7. `.pre-commit-config.yaml` (Pre-commit Configuration):** ```yaml repos: - repo: https://github.com/pre-commit/pre-commit-hooks rev: v4.4.0 hooks: - id: trailing-whitespace - id: end-of-file-fixer - id: check-yaml - id: check-added-large-files - repo: https://github.com/psf/black rev: 23.3.0 hooks: - id: black - repo: https://github.com/PyCQA/flake8 rev: 6.0.0 hooks: - id: flake8 - repo: https://github.com/pre-commit/mirrors-mypy rev: v1.0.0 hooks: - id: mypy additional_dependencies: - types-requests ``` **Explanation:** * This file configures pre-commit hooks that will run automatically before you commit code. * It includes hooks for: * Removing trailing whitespace. * Ensuring files end with a newline. * Checking YAML syntax. * Preventing large files from being added. * Formatting code with Black. * Checking code style with Flake8. * Running static type checking with MyPy. **Install pre-commit:** ```bash poetry run pip install pre-commit ``` **Install pre-commit hooks:** ```bash poetry run pre-commit install ``` Now, pre-commit hooks will run automatically whenever you try to commit code. **8. `README.md` (Project Readme):** ```markdown # Weather Service A simple weather service that provides weather data via HTTP and MCP. ## Requirements * Python 3.9+ * Poetry ## Installation ```bash poetry install ``` ## Running the Service ```bash poetry run uvicorn src.weather_service.main:app --reload ``` ## Running Tests ```bash poetry run pytest ``` ``` **Key Improvements and Considerations:** * **MCP Library Integration:** The most important part is the integration with your MCP library. You *must* replace the placeholder `mcp_lib` with the actual name of your library and adapt the `mcp_server()` and `handle_weather_request()` functions to use the library's API. This includes how you start the server, how you send and receive messages, and the format of the messages. * **Error Handling:** The example includes basic error handling (e.g., checking for the `city` in the request). You should add more robust error handling to handle potential issues with the MCP connection, data validation, and external API calls. * **Configuration:** Use environment variables or a configuration file (e.g., using Pydantic's `BaseSettings`) to manage configuration options like the MCP server address, port, and API keys. * **Logging:** Add logging to track requests, errors, and other important events. * **Asynchronous Operations:** The example uses `asyncio` and `async` functions to handle asynchronous operations. This is important for handling concurrent requests efficiently. * **Data Validation:** Use Pydantic to validate the incoming MCP requests and ensure that the data is in the correct format. * **Testing:** Write more comprehensive tests to cover different scenarios, including error cases and edge cases. Consider using mocking to isolate the weather logic from external dependencies. * **Deployment:** Consider how you will deploy the service (e.g., using Docker, Kubernetes, or a cloud platform). * **Security:** If your service handles sensitive data, implement appropriate security measures, such as authentication and authorization. * **Monitoring:** Implement monitoring to track the health and performance of the service. * **MCP Message Format:** The example assumes a simple dictionary format for MCP messages. You may need to adapt the code to handle a different format, such as JSON or Protocol Buffers. * **MCP Library Features:** Explore the features of your MCP library, such as message routing, error handling, and security. **How to Use This Example:** 1. **Replace Placeholders:** Replace the placeholder values (e.g., `Your Name`, `your.email@example.com`, `mcp-lib`, MCP server address and port) with your actual values. 2. **Install Dependencies:** Run `poetry install` to install the dependencies. 3. **Implement MCP Integration:** Adapt the `mcp_server()` and `handle_weather_request()` functions to use your MCP library. 4. **Run the Service:** Run `poetry run uvicorn src.weather_service.main:app --reload` to start the service. 5. **Test the Service:** Run `poetry run pytest` to run the tests. 6. **Commit Changes:** Commit your changes to trigger the pre-commit hooks. This comprehensive example provides a solid foundation for building a weather service with MCP integration, testing, and pre-commit setup. Remember to adapt the code to your specific needs and to follow best practices for software development. **Indonesian Translation:** Berikut adalah contoh implementasi server MCP (Microservice Communication Protocol) minimal dalam Python untuk data cuaca, beserta pengaturan pengujian dan pre-commit dasar. Ini adalah contoh yang disederhanakan untuk mengilustrasikan konsep inti. Anda mungkin perlu menyesuaikannya dengan kebutuhan spesifik Anda. **1. Struktur Proyek:** ``` weather_service/ ├── pyproject.toml ├── src/ │ └── weather_service/ │ ├── __init__.py │ ├── main.py │ ├── weather.py │ └── models.py ├── tests/ │ └── test_weather.py ├── .pre-commit-config.yaml └── README.md ``` **2. `pyproject.toml` (Konfigurasi Poetry):** ```toml [tool.poetry] name = "weather-service" version = "0.1.0" description = "Layanan cuaca sederhana menggunakan MCP" authors = ["Nama Anda <email.anda@contoh.com>"] readme = "README.md" [tool.poetry.dependencies] python = "^3.9" fastapi = "^0.100.0" uvicorn = "^0.23.0" pydantic = "^2.0" mcp-lib = "^0.1.0" # Ganti dengan library MCP Anda yang sebenarnya [tool.poetry.dev-dependencies] pytest = "^7.0" pytest-cov = "^4.0" pre-commit = "^3.0" flake8 = "^6.0" mypy = "^1.0" [build-system] requires = ["poetry-core"] build-backend = "poetry.core.masonry.api" [tool.pytest.ini_options] testpaths = ["tests"] addopts = "--cov=src/weather_service --cov-report term-missing" [tool.mypy] strict = true ignore_missing_imports = true ``` **Penjelasan:** * **`[tool.poetry]`:** Mendefinisikan metadata proyek. * **`[tool.poetry.dependencies]`:** Daftar dependensi runtime: * `fastapi`: Framework web modern dan cepat (kinerja tinggi) untuk membangun API. * `uvicorn`: Server ASGI (Asynchronous Server Gateway Interface) untuk menjalankan aplikasi FastAPI. * `pydantic`: Validasi data dan manajemen pengaturan menggunakan anotasi tipe Python. * `mcp-lib`: **Penting, ganti ini dengan nama library MCP Anda yang sebenarnya.** Ini adalah library yang menangani komunikasi MCP. Saya berasumsi Anda memiliki library yang menyediakan fungsi yang diperlukan untuk mengirim dan menerima pesan MCP. * **`[tool.poetry.dev-dependencies]`:** Daftar dependensi pengembangan: * `pytest`: Framework pengujian. * `pytest-cov`: Plugin pytest untuk pelaporan cakupan (coverage). * `pre-commit`: Framework untuk mengelola hook pre-commit. * `flake8`: Pemeriksa gaya kode. * `mypy`: Pemeriksa tipe statis. * **`[tool.pytest.ini_options]`:** Mengkonfigurasi pytest. * **`[tool.mypy]`:** Mengkonfigurasi mypy. **Instal dependensi:** ```bash poetry install ``` **3. `src/weather_service/models.py` (Model Data):** ```python from pydantic import BaseModel class WeatherData(BaseModel): city: str temperature: float condition: str ``` **Penjelasan:** * Menggunakan Pydantic untuk mendefinisikan model `WeatherData`. Model ini akan digunakan untuk menyusun data cuaca yang dikembalikan oleh layanan. **4. `src/weather_service/weather.py` (Logika Cuaca):** ```python from .models import WeatherData import random def get_weather(city: str) -> WeatherData: """ Mensimulasikan pengambilan data cuaca untuk kota tertentu. """ temperature = random.uniform(10, 35) # Mensimulasikan suhu antara 10 dan 35 conditions = ["Sunny", "Cloudy", "Rainy", "Snowy"] condition = random.choice(conditions) return WeatherData(city=city, temperature=temperature, condition=condition) ``` **Penjelasan:** * `get_weather(city: str)`: Fungsi ini mensimulasikan pengambilan data cuaca. Dalam aplikasi nyata, Anda akan mengganti ini dengan panggilan ke API cuaca eksternal atau database. Ini mengembalikan objek `WeatherData`. **5. `src/weather_service/main.py` (Aplikasi FastAPI dan Integrasi MCP):** ```python from fastapi import FastAPI from .weather import get_weather from .models import WeatherData import asyncio import mcp_lib # Ganti dengan library MCP Anda yang sebenarnya app = FastAPI() # Konfigurasi MCP (Ganti dengan konfigurasi Anda yang sebenarnya) MCP_SERVER_ADDRESS = "localhost" MCP_SERVER_PORT = 5000 async def handle_weather_request(request_data: dict) -> dict: """ Menangani permintaan cuaca yang masuk melalui MCP. """ city = request_data.get("city") if not city: return {"error": "Kota diperlukan"} weather_data = get_weather(city) return weather_data.dict() # Konversi ke dictionary untuk MCP async def mcp_server(): """ Memulai server MCP dan mendengarkan permintaan cuaca. """ async def message_handler(message: dict) -> dict: """Menangani pesan MCP yang masuk.""" print(f"Menerima pesan MCP: {message}") return await handle_weather_request(message) await mcp_lib.start_server(MCP_SERVER_ADDRESS, MCP_SERVER_PORT, message_handler) # Ganti dengan fungsi memulai server MCP Anda @app.on_event("startup") async def startup_event(): """Memulai server MCP saat aplikasi dimulai.""" asyncio.create_task(mcp_server()) @app.get("/weather/{city}") async def get_weather_http(city: str) -> WeatherData: """ Endpoint HTTP untuk mendapatkan data cuaca. """ return get_weather(city) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000) ``` **Penjelasan:** * **Pengaturan FastAPI:** * Membuat instance aplikasi FastAPI. * Mendefinisikan endpoint HTTP `/weather/{city}` yang memanggil fungsi `get_weather` dan mengembalikan hasilnya. * **Integrasi MCP:** * `MCP_SERVER_ADDRESS` dan `MCP_SERVER_PORT`: Konfigurasi untuk server MCP. **Ganti ini dengan konfigurasi Anda yang sebenarnya.** * `handle_weather_request(request_data: dict)`: Fungsi ini adalah inti dari integrasi MCP. Ini menerima dictionary yang mewakili permintaan MCP, mengekstrak `city` dari permintaan, memanggil fungsi `get_weather`, dan mengembalikan data cuaca sebagai dictionary. **Di sinilah Anda akan menyesuaikan kode untuk menangani format spesifik pesan MCP Anda.** * `mcp_server()`: Fungsi ini memulai server MCP. **Anda perlu mengganti `mcp_lib.start_server()` dengan fungsi sebenarnya dari library MCP Anda yang memulai server.** Fungsi `message_handler` dipanggil setiap kali pesan MCP diterima. * `startup_event()`: Fungsi ini dipanggil saat aplikasi FastAPI dimulai. Ini memulai server MCP di tugas latar belakang menggunakan `asyncio.create_task()`. * **Penting:** Contoh ini mengasumsikan bahwa library MCP Anda menyediakan fungsi seperti `mcp_lib.start_server()` yang mengambil alamat, port, dan fungsi penanganan pesan sebagai argumen. Anda perlu menyesuaikan kode ini agar sesuai dengan API library MCP Anda. Fungsi `message_handler` harus mengambil pesan MCP yang masuk sebagai input dan mengembalikan respons. **6. `tests/test_weather.py` (Pengujian):** ```python from src.weather_service.weather import get_weather from src.weather_service.models import WeatherData def test_get_weather(): weather_data = get_weather("London") assert isinstance(weather_data, WeatherData) assert weather_data.city == "London" assert isinstance(weather_data.temperature, float) assert isinstance(weather_data.condition, str) ``` **Penjelasan:** * Kasus pengujian sederhana yang memeriksa apakah fungsi `get_weather` mengembalikan objek `WeatherData` dengan kota dan tipe data yang benar. **Jalankan pengujian:** ```bash poetry run pytest ``` **7. `.pre-commit-config.yaml` (Konfigurasi Pre-commit):** ```yaml repos: - repo: https://github.com/pre-commit/pre-commit-hooks rev: v4.4.0 hooks: - id: trailing-whitespace - id: end-of-file-fixer - id: check-yaml - id: check-added-large-files - repo: https://github.com/psf/black rev: 23.3.0 hooks: - id: black - repo: https://github.com/PyCQA/flake8 rev: 6.0.0 hooks: - id: flake8 - repo: https://github.com/pre-commit/mirrors-mypy rev: v1.0.0 hooks: - id: mypy additional_dependencies: - types-requests ``` **Penjelasan:** * File ini mengkonfigurasi hook pre-commit yang akan berjalan secara otomatis sebelum Anda melakukan commit kode. * Ini mencakup hook untuk: * Menghapus spasi kosong di akhir baris. * Memastikan file diakhiri dengan baris baru. * Memeriksa sintaks YAML. * Mencegah penambahan file besar. * Memformat kode dengan Black. * Memeriksa gaya kode dengan Flake8. * Menjalankan pemeriksaan tipe statis dengan MyPy. **Instal pre-commit:** ```bash poetry run pip install pre-commit ``` **Instal hook pre-commit:** ```bash poetry run pre-commit install ``` Sekarang, hook pre-commit akan berjalan secara otomatis setiap kali Anda mencoba melakukan commit kode. **8. `README.md` (Readme Proyek):** ```markdown # Layanan Cuaca Layanan cuaca sederhana yang menyediakan data cuaca melalui HTTP dan MCP. ## Persyaratan * Python 3.9+ * Poetry ## Instalasi ```bash poetry install ``` ## Menjalankan Layanan ```bash poetry run uvicorn src.weather_service.main:app --reload ``` ## Menjalankan Pengujian ```bash poetry run pytest ``` ``` **Peningkatan dan Pertimbangan Utama:** * **Integrasi Library MCP:** Bagian terpenting adalah integrasi dengan library MCP Anda. Anda *harus* mengganti placeholder `mcp_lib` dengan nama library Anda yang sebenarnya dan menyesuaikan fungsi `mcp_server()` dan `handle_weather_request()` untuk menggunakan API library. Ini termasuk cara Anda memulai server, cara Anda mengirim dan menerima pesan, dan format pesan. * **Penanganan Kesalahan:** Contoh ini mencakup penanganan kesalahan dasar (misalnya, memeriksa `city` dalam permintaan). Anda harus menambahkan penanganan kesalahan yang lebih kuat untuk menangani potensi masalah dengan koneksi MCP, validasi data, dan panggilan API eksternal. * **Konfigurasi:** Gunakan variabel lingkungan atau file konfigurasi (misalnya, menggunakan `BaseSettings` Pydantic) untuk mengelola opsi konfigurasi seperti alamat server MCP, port, dan kunci API. * **Pencatatan Log (Logging):** Tambahkan pencatatan log untuk melacak permintaan, kesalahan, dan peristiwa penting lainnya. * **Operasi Asinkron:** Contoh ini menggunakan `asyncio` dan fungsi `async` untuk menangani operasi asinkron. Ini penting untuk menangani permintaan bersamaan secara efisien. * **Validasi Data:** Gunakan Pydantic untuk memvalidasi permintaan MCP yang masuk dan memastikan bahwa data dalam format yang benar. * **Pengujian:** Tulis pengujian yang lebih komprehensif untuk mencakup skenario yang berbeda, termasuk kasus kesalahan dan kasus tepi. Pertimbangkan untuk menggunakan mocking untuk mengisolasi logika cuaca dari dependensi eksternal. * **Penyebaran (Deployment):** Pertimbangkan bagaimana Anda akan menyebarkan layanan (misalnya, menggunakan Docker, Kubernetes, atau platform cloud). * **Keamanan:** Jika layanan Anda menangani data sensitif, terapkan langkah-langkah keamanan yang sesuai, seperti otentikasi dan otorisasi. * **Pemantauan (Monitoring):** Terapkan pemantauan untuk melacak kesehatan dan kinerja layanan. * **Format Pesan MCP:** Contoh ini mengasumsikan format dictionary sederhana untuk pesan MCP. Anda mungkin perlu menyesuaikan kode untuk menangani format yang berbeda, seperti JSON atau Protocol Buffers. * **Fitur Library MCP:** Jelajahi fitur library MCP Anda, seperti perutean pesan, penanganan kesalahan, dan keamanan. **Cara Menggunakan Contoh Ini:** 1. **Ganti Placeholder:** Ganti nilai placeholder (misalnya, `Nama Anda`, `email.anda@contoh.com`, `mcp-lib`, alamat dan port server MCP) dengan nilai Anda yang sebenarnya. 2. **Instal Dependensi:** Jalankan `poetry install` untuk menginstal dependensi. 3. **Implementasikan Integrasi MCP:** Sesuaikan fungsi `mcp_server()` dan `handle_weather_request()` untuk menggunakan library MCP Anda. 4. **Jalankan Layanan:** Jalankan `poetry run uvicorn src.weather_service.main:app --reload` untuk memulai layanan. 5. **Uji Layanan:** Jalankan `poetry run pytest` untuk menjalankan pengujian. 6. **Commit Perubahan:** Commit perubahan Anda untuk memicu hook pre-commit. Contoh komprehensif ini memberikan dasar yang kuat untuk membangun layanan cuaca dengan integrasi MCP, pengujian, dan pengaturan pre-commit. Ingatlah untuk menyesuaikan kode dengan kebutuhan spesifik Anda dan untuk mengikuti praktik terbaik untuk pengembangan perangkat lunak.

Playcanvas_editor Mcp Server

Cermin dari

mcp-tools-cli

command-line client for interacting with Model Context Protocol (MCP) servers.

IACR Cryptology ePrint Archive MCP Server

Cermin dari

MCP Server Playground

Mirror of

Grasshopper MCP サーバー

Implementasi server Model Context Protocol (MCP) untuk integrasi Rhinoceros/Grasshopper, memungkinkan model AI berinteraksi dengan alat desain parametrik.

Polkassembly MCP Server

MCP Server for Polkassembly API

langchain-mcp

Model Context Protocol tool support for LangChain

MailPace MCP (Model Context Protocol) Server

Implementasi server MCP yang memungkinkan pengiriman email melalui API email transaksional cepat MailPace.

mcptime

SImple MCP server to return the current time

RapidAPI MCP Server

Okay, here's a breakdown of how you might implement an MCP (presumably referring to a **Microservice Control Plane** or a similar architecture) server for integrating the RapidAPI Global Patent API with SQLite storage, along with considerations and code snippets (in Python, as it's a common choice for microservices and API interactions). **Conceptual Architecture** 1. **Client (Your Application):** Sends requests (e.g., patent search queries) to the MCP server. 2. **MCP Server (This is what we're building):** * **API Gateway/Router:** Receives requests from the client. Handles authentication, authorization, rate limiting, and routing. * **Patent Service:** The core logic for interacting with the RapidAPI Global Patent API. * **Data Storage Service:** Handles interactions with the SQLite database. 3. **RapidAPI Global Patent API:** The external API providing patent data. 4. **SQLite Database:** Stores patent data (e.g., search results, cached data). **Key Components and Implementation Steps** **1. Project Setup (Python with Flask/FastAPI)** * **Choose a Framework:** Flask and FastAPI are popular choices for building microservices in Python. FastAPI is generally preferred for its performance and built-in features like automatic API documentation (Swagger/OpenAPI). I'll use FastAPI in this example. * **Install Dependencies:** ```bash pip install fastapi uvicorn requests python-dotenv pysqlite3 ``` * `fastapi`: The FastAPI framework. * `uvicorn`: An ASGI server to run the FastAPI application. * `requests`: For making HTTP requests to the RapidAPI. * `python-dotenv`: For loading environment variables from a `.env` file. * `pysqlite3`: SQLite driver for Python. * **Project Structure:** ``` patent_mcp/ ├── main.py # Main application entry point ├── services/ │ ├── patent_service.py # Logic for interacting with RapidAPI │ └── data_service.py # Logic for interacting with SQLite ├── models/ │ └── patent_model.py # Data models (Pydantic) ├── .env # Environment variables (API keys, etc.) └── README.md ``` **2. Environment Variables (.env)** Create a `.env` file in the root directory: ``` RAPIDAPI_KEY="YOUR_RAPIDAPI_KEY" RAPIDAPI_HOST="global-patent.p.rapidapi.com" DATABASE_PATH="patents.db" ``` **3. Data Models (models/patent_model.py)** Use Pydantic to define data models for patent information. This helps with data validation and serialization. ```python from pydantic import BaseModel from typing import Optional, List class Patent(BaseModel): patent_id: str title: str abstract: Optional[str] = None inventors: Optional[List[str]] = None assignees: Optional[List[str]] = None publication_date: Optional[str] = None # Add other relevant fields from the RapidAPI response ``` **4. Data Storage Service (services/data_service.py)** This module handles interactions with the SQLite database. ```python import sqlite3 from typing import List from models.patent_model import Patent import os from dotenv import load_dotenv load_dotenv() DATABASE_PATH = os.getenv("DATABASE_PATH") def create_connection(): """Creates a database connection to the SQLite database.""" conn = None try: conn = sqlite3.connect(DATABASE_PATH) return conn except sqlite3.Error as e: print(f"Error connecting to database: {e}") return None def create_table(): """Creates the patents table if it doesn't exist.""" conn = create_connection() if conn: try: cursor = conn.cursor() cursor.execute(""" CREATE TABLE IF NOT EXISTS patents ( patent_id TEXT PRIMARY KEY, title TEXT, abstract TEXT, inventors TEXT, assignees TEXT, publication_date TEXT ) """) conn.commit() except sqlite3.Error as e: print(f"Error creating table: {e}") finally: conn.close() def insert_patent(patent: Patent): """Inserts a patent into the database.""" conn = create_connection() if conn: try: cursor = conn.cursor() cursor.execute(""" INSERT INTO patents (patent_id, title, abstract, inventors, assignees, publication_date) VALUES (?, ?, ?, ?, ?, ?) """, (patent.patent_id, patent.title, patent.abstract, ','.join(patent.inventors) if patent.inventors else None, ','.join(patent.assignees) if patent.assignees else None, patent.publication_date)) conn.commit() except sqlite3.Error as e: print(f"Error inserting patent: {e}") finally: conn.close() def get_patent(patent_id: str) -> Patent | None: """Retrieves a patent from the database by patent_id.""" conn = create_connection() if conn: try: cursor = conn.cursor() cursor.execute("SELECT patent_id, title, abstract, inventors, assignees, publication_date FROM patents WHERE patent_id=?", (patent_id,)) row = cursor.fetchone() if row: patent = Patent( patent_id=row[0], title=row[1], abstract=row[2], inventors=row[3].split(',') if row[3] else [], assignees=row[4].split(',') if row[4] else [], publication_date=row[5] ) return patent else: return None except sqlite3.Error as e: print(f"Error retrieving patent: {e}") return None finally: conn.close() def search_patents(query: str) -> List[Patent]: """Searches for patents in the database based on a query (e.g., in title or abstract).""" conn = create_connection() patents: List[Patent] = [] if conn: try: cursor = conn.cursor() # Simple example: Search title and abstract. Use FTS (Full-Text Search) for better performance in a real application. cursor.execute("SELECT patent_id, title, abstract, inventors, assignees, publication_date FROM patents WHERE title LIKE ? OR abstract LIKE ?", ('%' + query + '%', '%' + query + '%')) rows = cursor.fetchall() for row in rows: patent = Patent( patent_id=row[0], title=row[1], abstract=row[2], inventors=row[3].split(',') if row[3] else [], assignees=row[4].split(',') if row[4] else [], publication_date=row[5] ) patents.append(patent) except sqlite3.Error as e: print(f"Error searching patents: {e}") finally: conn.close() return patents # Initialize the database table when the module is loaded create_table() ``` **5. Patent Service (services/patent_service.py)** This module interacts with the RapidAPI Global Patent API. ```python import requests import os from dotenv import load_dotenv from typing import Optional, Dict, Any from models.patent_model import Patent import services.data_service as data_service load_dotenv() RAPIDAPI_KEY = os.getenv("RAPIDAPI_KEY") RAPIDAPI_HOST = os.getenv("RAPIDAPI_HOST") def search_patents_rapidapi(query: str) -> Optional[Dict[str, Any]]: """Searches the RapidAPI Global Patent API.""" url = "https://global-patent.p.rapidapi.com/search" # Replace with the correct endpoint headers = { "X-RapidAPI-Key": RAPIDAPI_KEY, "X-RapidAPI-Host": RAPIDAPI_HOST } querystring = {"q": query} try: response = requests.get(url, headers=headers, params=querystring) response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx) return response.json() except requests.exceptions.RequestException as e: print(f"Error calling RapidAPI: {e}") return None def get_patent_details(patent_id: str) -> Optional[Patent]: """Retrieves patent details from RapidAPI or the database (with caching).""" # 1. Check the database first patent = data_service.get_patent(patent_id) if patent: print(f"Patent {patent_id} found in database.") return patent # 2. If not in the database, fetch from RapidAPI url = f"https://global-patent.p.rapidapi.com/patent/{patent_id}" # Replace with the correct endpoint headers = { "X-RapidAPI-Key": RAPIDAPI_KEY, "X-RapidAPI-Host": RAPIDAPI_HOST } try: response = requests.get(url, headers=headers) response.raise_for_status() data = response.json() # Assuming the RapidAPI response structure, adapt this part patent = Patent( patent_id=data.get("patent_id"), title=data.get("title"), abstract=data.get("abstract"), inventors=data.get("inventors"), assignees=data.get("assignees"), publication_date=data.get("publication_date") ) # 3. Store the fetched patent in the database for caching data_service.insert_patent(patent) print(f"Patent {patent_id} fetched from RapidAPI and stored in database.") return patent except requests.exceptions.RequestException as e: print(f"Error calling RapidAPI: {e}") return None except Exception as e: print(f"Error processing RapidAPI response: {e}") return None ``` **6. Main Application (main.py)** This is the entry point for your FastAPI application. ```python from fastapi import FastAPI, HTTPException, Query from typing import List, Optional from models.patent_model import Patent import services.patent_service as patent_service import services.data_service as data_service app = FastAPI() @app.get("/patents/search", response_model=List[Patent]) async def search_patents(query: str = Query(..., title="Search Query", description="The search term for patents.")): """Searches for patents, first in the local database, then in the RapidAPI if not found locally.""" # 1. Search in the local database local_results = data_service.search_patents(query) if local_results: print(f"Found {len(local_results)} patents in local database for query: {query}") return local_results # 2. If no results in the database, search the RapidAPI rapidapi_results = patent_service.search_patents_rapidapi(query) if rapidapi_results and isinstance(rapidapi_results, dict) and "results" in rapidapi_results: # Adjust based on actual API response patents: List[Patent] = [] for result in rapidapi_results["results"]: # Adjust based on actual API response try: patent = Patent( patent_id=result.get("patent_id"), title=result.get("title"), abstract=result.get("abstract"), inventors=result.get("inventors"), assignees=result.get("assignees"), publication_date=result.get("publication_date") ) data_service.insert_patent(patent) # Cache the result patents.append(patent) except Exception as e: print(f"Error processing RapidAPI result: {e}") print(f"Found {len(patents)} patents in RapidAPI for query: {query}") return patents else: print(f"No patents found in RapidAPI for query: {query}") return [] # Return an empty list if no results are found @app.get("/patents/{patent_id}", response_model=Patent) async def get_patent(patent_id: str): """Retrieves a specific patent by its ID, using caching.""" patent = patent_service.get_patent_details(patent_id) if patent: return patent else: raise HTTPException(status_code=404, detail="Patent not found") @app.get("/health") async def health_check(): """Health check endpoint.""" return {"status": "ok"} ``` **7. Running the Application** ```bash uvicorn main:app --reload ``` This will start the FastAPI server. You can then access the API endpoints (e.g., `http://127.0.0.1:8000/patents/search?query=example`, `http://127.0.0.1:8000/patents/US1234567B1`). FastAPI will automatically generate API documentation at `http://127.0.0.1:8000/docs`. **Important Considerations and Enhancements** * **Error Handling:** Implement more robust error handling, including logging and appropriate HTTP status codes. * **API Key Security:** Never hardcode API keys directly into your code. Use environment variables and secure configuration management. Consider using a secrets management service (e.g., HashiCorp Vault) for production environments. * **Rate Limiting:** Implement rate limiting to prevent abuse of your API and to stay within the RapidAPI usage limits. FastAPI has middleware options for rate limiting. * **Authentication/Authorization:** Implement authentication and authorization to control access to your API. OAuth 2.0 is a common choice. * **Asynchronous Operations:** Use `async` and `await` for I/O-bound operations (like API calls and database interactions) to improve performance. FastAPI is built for asynchronous programming. * **Data Validation:** Use Pydantic models for strict data validation. * **Testing:** Write unit tests and integration tests to ensure the reliability of your code. * **Logging:** Implement comprehensive logging to track API usage, errors, and performance. * **Caching:** Implement more sophisticated caching strategies (e.g., using Redis or Memcached) for frequently accessed data. Consider using a cache invalidation strategy. * **Full-Text Search (FTS):** For more efficient searching of patent data, use SQLite's FTS extension (Full-Text Search). This will significantly improve search performance, especially for large datasets. You'll need to create a virtual table using FTS and populate it with the text data you want to search. * **Database Migrations:** Use a database migration tool (e.g., Alembic) to manage database schema changes. * **Deployment:** Consider deploying your microservice to a container orchestration platform like Kubernetes or Docker Swarm. * **Monitoring:** Implement monitoring to track the health and performance of your microservice. Tools like Prometheus and Grafana can be used for monitoring. * **Circuit Breaker:** Implement a circuit breaker pattern to prevent cascading failures if the RapidAPI becomes unavailable. * **API Gateway:** For a true MCP, use a dedicated API Gateway (e.g., Kong, Tyk, Ambassador) to handle routing, authentication, rate limiting, and other cross-cutting concerns. This decouples these concerns from your individual microservices. * **Service Discovery:** In a microservices architecture, use a service discovery mechanism (e.g., Consul, etcd, Kubernetes DNS) to allow your services to find each other. * **Message Queue:** For asynchronous tasks (e.g., updating the database with new patent data), use a message queue (e.g., RabbitMQ, Kafka). **Example of using FTS (Full-Text Search) in SQLite** First, enable FTS5 in your SQLite database. You'll need to modify the `create_table` function in `data_service.py`: ```python def create_table(): """Creates the patents table with FTS5 if it doesn't exist.""" conn = create_connection() if conn: try: cursor = conn.cursor() cursor.execute("DROP TABLE IF EXISTS patents;") # Drop the old table cursor.execute(""" CREATE VIRTUAL TABLE IF NOT EXISTS patents USING fts5( patent_id, title, abstract, inventors, assignees, publication_date, content='tokenize=porter' -- Use Porter Stemmer for better search ); """) conn.commit() except sqlite3.Error as e: print(f"Error creating table: {e}") finally: conn.close() ``` Then, modify the `insert_patent` function to insert data into the FTS table: ```python def insert_patent(patent: Patent): """Inserts a patent into the FTS table.""" conn = create_connection() if conn: try: cursor = conn.cursor() cursor.execute(""" INSERT INTO patents (patent_id, title, abstract, inventors, assignees, publication_date) VALUES (?, ?, ?, ?, ?, ?) """, (patent.patent_id, patent.title, patent.abstract, ','.join(patent.inventors) if patent.inventors else None, ','.join(patent.assignees) if patent.assignees else None, patent.publication_date)) conn.commit() except sqlite3.Error as e: print(f"Error inserting patent: {e}") finally: conn.close() ``` Finally, update the `search_patents` function to use the FTS `MATCH` operator: ```python def search_patents(query: str) -> List[Patent]: """Searches for patents in the database using FTS.""" conn = create_connection() patents: List[Patent] = [] if conn: try: cursor = conn.cursor() cursor.execute(""" SELECT patent_id, title, abstract, inventors, assignees, publication_date FROM patents WHERE patents MATCH ? """, (query,)) # Use the MATCH operator for FTS rows = cursor.fetchall() for row in rows: patent = Patent( patent_id=row[0], title=row[1], abstract=row[2], inventors=row[3].split(',') if row[3] else [], assignees=row[4].split(',') if row[4] else [], publication_date=row[5] ) patents.append(patent) except sqlite3.Error as e: print(f"Error searching patents: {e}") finally: conn.close() return patents ``` **Explanation of the FTS Changes:** * **`CREATE VIRTUAL TABLE ... USING fts5(...)`:** Creates a virtual table using the `fts5` module. The `content='tokenize=porter'` option enables the Porter Stemmer for better search results (e.g., searching for "running" will also find "run"). * **`patents MATCH ?`:** Uses the `MATCH` operator to perform a full-text search. The `?` placeholder is replaced with the search query. Remember to drop the existing `patents` table before creating the FTS table. Also, you might need to install the `sqlite-fts5` extension separately, depending on your system. This comprehensive guide provides a solid foundation for building your MCP server. Remember to adapt the code and architecture to your specific needs and the specifics of the RapidAPI Global Patent API response format. Good luck!

mcp-sondehub

Sebuah Server MCP untuk API SondeHub terutama untuk bereksperimen dengan MCP.

JetBrains MCP Server Plugin

JetBrains MCP Server Plugin

Audio MCP Server

Memungkinkan Claude dan asisten AI lainnya untuk berinteraksi dengan sistem audio komputer Anda, memungkinkan perekaman dari mikrofon dan pemutaran audio melalui speaker.

Office-PowerPoint-MCP-Server

A server that enables creating and editing PowerPoint presentations programmatically through the Model Context Protocol, supporting features like adding slides, images, textboxes, charts, and tables.

Figma MCP Server

Server Protokol Konteks Model yang terintegrasi dengan API Figma, memungkinkan interaksi dengan file Figma, komentar, komponen, proyek, dan manajemen webhook.

OpenAI Speech-to-Text transcriptions MCP Server

Sebuah server MCP yang memungkinkan transkripsi berkas audio menggunakan API Speech-to-Text OpenAI, dengan dukungan untuk berbagai bahasa dan opsi penyimpanan berkas.

FastDomainCheck MCP Server

Protokol Konteks Model untuk memeriksa status pendaftaran nama domain secara massal.

Claude Desktop Notion MCP Server

Server Protokol Konteks Model sistem berkas yang menyediakan Claude Desktop dengan kemampuan untuk membaca, menulis, dan memanipulasi berkas di sistem Anda.

Medrxiv

🔍 Aktifkan asisten AI untuk mencari dan mengakses makalah medRxiv melalui antarmuka MCP yang sederhana. Server MCP medRxiv menyediakan jembatan antara asisten AI dan repositori pracetak medRxiv melalui Model Context Protocol (MCP). Ini memungkinkan model AI untuk mencari pracetak ilmu kesehatan dan mengaksesnya.

MCP Screenshot Server

Mirror of

Gemini Image Generation MCP Server

Server Protokol Konteks Model yang menyediakan kemampuan pembuatan gambar menggunakan Google Gemini 2 API, memungkinkan pengguna untuk menghasilkan banyak gambar dengan parameter yang dapat disesuaikan seperti prompt, rasio aspek, dan pengaturan pembuatan orang.

Teamwork MCP

Sebuah server MCP yang terhubung ke Teamwork API, menyediakan antarmuka yang disederhanakan untuk berinteraksi dengan proyek dan tugas Teamwork.

Mem0

Jembatan antara aplikasi MCP Host dan layanan cloud mem0, yang dikhususkan untuk manajemen proyek dengan kemampuan untuk menyimpan, mengambil, dan mencari informasi proyek dalam format terstruktur.

MCP Todoist

Server Protokol Konteks Model yang memungkinkan Claude untuk berinteraksi dengan akun Todoist Anda, memungkinkan Anda untuk mengelola tugas, proyek, dan label melalui bahasa alami.

Solana MCP Server

Server MCP yang memungkinkan untuk melakukan tindakan nyata terkait Solana. Versi pertama menambahkan pengetahuan tentang metode RPC dan cara memanggilnya.

IOD App

discover and install MCP servers

Sentry MCP Server

Server Protokol Konteks Model yang memungkinkan asisten AI berinteraksi dengan Sentry API untuk mengambil dan menganalisis data kesalahan, mengelola proyek, dan memantau kinerja aplikasi.