Discover Awesome MCP Servers

HQ Pool Services Website

Test w/ Figma MCP server to generate a pool services page

🌟 Unsplash MCP Server Repository

🔎 A MCP server for Unsplash image search.

MQTTX SSE Server

Uma implementação do Protocolo Modelo-Contexto (MCP) que permite operações MQTT sobre o transporte de Eventos Enviados pelo Servidor (SSE).

mcp-server

learn to make some mcp servers

Minimal MCP Server

A minimal implementation of a Model Context Protocol (MCP) server

🦜 🔗 LangChain MCP Client

🦜🔗 LangChain Model Context Protocol (MCP) Client

@enemyrr/mcp-mysql-server

Espelho de

MCP Actions Adapter

A simple adapter to convert a MCP server to a GPT actions compatible API

worker17

An MCP server to monitor workers productivity and fire them as needed.

Mcp Servers Collection

Coleção de servidores e integrações MCP verificados

Wisdom MCP Gateway

A stdio gateway for the Enterpret's Wisdom MCP SSE Server

Local iMessage RAG MCP Server

iMessage RAG MCP Server from Anthropic MCP Hackathon (NYC)

Mcp Namecheap Registrar

Connects to namecheap api for checking availability and pricing of domains and registering them.

T2_C2

Server code for MCP

MCP Bundler Service

Um microsserviço para agrupar servidores MCP de repositórios GitHub e prepará-los para implantação.

mcp-server-yahoo-finance MCP server

Espelho de

MCP Test Client

MCP Test Client is a TypeScript testing utility for Model Context Protocol (MCP) servers.

ActionKit MCP Starter

Here's a basic starter code structure for an MCP (Minecraft Protocol) server powered by ActionKit, along with explanations and considerations. This is a simplified example and will require significant expansion to handle the full Minecraft protocol. ```python import asyncio import struct import logging from actionkit.action import Action, ActionContext # Configure logging (important for debugging) logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # --- ActionKit Integration --- class MinecraftAction(Action): """ Base class for Minecraft-related actions. Provides common functionality. """ def __init__(self, context: ActionContext, client_socket: asyncio.StreamWriter): super().__init__(context) self.client_socket = client_socket async def send_packet(self, packet_id: int, data: bytes): """ Sends a Minecraft packet to the client. """ packet = struct.pack(">b", packet_id) + data # >b is big-endian signed byte length = len(packet) length_bytes = struct.pack(">i", length) # >i is big-endian signed integer await self.client_socket.write(length_bytes + packet) await self.client_socket.drain() # Ensure data is sent logger.debug(f"Sent packet ID {packet_id} with length {length}") class HandleHandshake(MinecraftAction): """ Handles the initial handshake from the client. """ async def execute(self, protocol_version: int, server_address: str, server_port: int, next_state: int): logger.info(f"Handshake received: protocol={protocol_version}, address={server_address}, port={server_port}, next_state={next_state}") if next_state == 1: # Status await HandleStatusRequest(self.context, self.client_socket).execute() elif next_state == 2: # Login await HandleLoginStart(self.context, self.client_socket).execute() else: logger.warning(f"Unknown next_state: {next_state}") await self.client_socket.close() class HandleStatusRequest(MinecraftAction): """ Handles the status request (ping). """ async def execute(self): logger.info("Status request received.") # Construct the server status JSON (example) status_json = """ { "version": { "name": "My Awesome Server", "protocol": 763 // Example protocol version }, "players": { "max": 100, "online": 0, "sample": [] }, "description": { "text": "A server powered by ActionKit!" } } """ status_bytes = status_json.encode("utf-8") length = len(status_bytes) length_bytes = struct.pack(">i", length) await self.send_packet(0x00, length_bytes + status_bytes) # Send a ping response (optional) # await self.send_packet(0x01, b"\x00\x00\x00\x00\x00\x00\x00\x00") # Example ping response # Close the connection after status await self.client_socket.close() class HandleLoginStart(MinecraftAction): """ Handles the login start packet. """ async def execute(self, username: str): logger.info(f"Login start received: username={username}") # TODO: Authentication logic here (e.g., Mojang authentication) # For now, just accept the login and send a success packet uuid = "00000000-0000-0000-0000-000000000000" # Replace with a real UUID username_bytes = username.encode("utf-8") uuid_bytes = uuid.encode("utf-8") data = len(uuid_bytes).to_bytes(1, 'big') + uuid_bytes + len(username_bytes).to_bytes(1, 'big') + username_bytes await self.send_packet(0x02, data) # Transition to play state (example) await HandleJoinGame(self.context, self.client_socket).execute() class HandleJoinGame(MinecraftAction): """ Handles the Join Game packet, sending initial world information. """ async def execute(self): logger.info("Sending Join Game packet.") # Example Join Game packet data (replace with your actual values) entity_id = 0 gamemode = 1 # Creative mode dimension = 0 # Overworld hashed_seed = 0 max_players = 100 level_type = "default".encode("utf-8") simulation_distance = 8 reduced_debug_info = False enable_respawn_screen = True is_debug = False is_flat = False data = struct.pack(">i", entity_id) # Entity ID data += struct.pack(">b", gamemode) # Gamemode data += struct.pack(">i", dimension) # Dimension data += struct.pack(">q", hashed_seed) # Hashed Seed data += struct.pack(">b", max_players) # Max Players data += len(level_type).to_bytes(1, 'big') + level_type # Level Type data += struct.pack(">i", simulation_distance) # Simulation Distance data += struct.pack("?", reduced_debug_info) # Reduced Debug Info data += struct.pack("?", enable_respawn_screen) # Enable Respawn Screen data += struct.pack("?", is_debug) # Is Debug data += struct.pack("?", is_flat) # Is Flat await self.send_packet(0x26, data) # Join Game Packet ID # Send other necessary packets (e.g., spawn position, player position) await self.send_packet(0x49, struct.pack(">qi", 0, 0)) # Spawn Position Packet ID # Example Player Position and Look packet x, y, z = 0.0, 64.0, 0.0 yaw, pitch = 0.0, 0.0 flags = 0x00 # No flags teleport_id = 0 data = struct.pack(">dddffbi", x, y, z, yaw, pitch, flags, teleport_id) await self.send_packet(0x36, data) # Player Position and Look Packet ID # Send Chunk Data (very complex, requires chunk generation) # This is a placeholder - you'll need to implement chunk generation # and the correct packet structure. # await self.send_packet(0x22, b"...") # Send Clientbound Plugin Message (example) channel = "minecraft:brand".encode("utf-8") data = "vanilla".encode("utf-8") packet_data = len(channel).to_bytes(1, 'big') + channel + len(data).to_bytes(1, 'big') + data await self.send_packet(0x17, packet_data) logger.info("Join Game sequence completed.") # --- Server Logic --- async def handle_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter, context: ActionContext): """ Handles a single client connection. """ addr = writer.get_extra_info('peername') logger.info(f"New connection from {addr}") try: # --- Handshake --- length_bytes = await reader.readexactly(4) length = struct.unpack(">i", length_bytes)[0] packet_bytes = await reader.readexactly(length) packet_id = struct.unpack(">b", packet_bytes[:1])[0] data = packet_bytes[1:] if packet_id == 0x00: # Handshake protocol_version, server_address_length = struct.unpack(">ib", data[:5]) server_address = data[5:5 + server_address_length].decode("utf-8") server_port, next_state = struct.unpack(">hi", data[5 + server_address_length:]) await HandleHandshake(context, writer).execute(protocol_version, server_address, server_port, next_state) elif packet_id == 0x00: # Legacy Ping # Handle legacy ping (before 1.7) logger.info("Legacy Ping received.") response = b"\xffServer\x00Awesome Server\x00Some Message\x001\x00100" await writer.write(response) await writer.drain() await writer.close() else: logger.warning(f"Unknown packet ID: {packet_id}") await writer.close() # --- Example Login (after handshake) --- # This part would be more complex in a real server, # involving reading more packets and handling player input. # Example: Read Login Start packet # length_bytes = await reader.readexactly(4) # length = struct.unpack(">i", length_bytes)[0] # packet_bytes = await reader.readexactly(length) # packet_id = struct.unpack(">b", packet_bytes[:1])[0] # if packet_id == 0x00: # username_length = struct.unpack(">b", packet_bytes[1:2])[0] # username = packet_bytes[2:2 + username_length].decode("utf-8") # await HandleLoginStart(context, writer).execute(username) except asyncio.IncompleteReadError: logger.warning(f"Client {addr} disconnected unexpectedly.") except Exception as e: logger.exception(f"Error handling client {addr}: {e}") finally: writer.close() await writer.wait_closed() logger.info(f"Connection from {addr} closed.") async def main(): """ Main server function. """ context = ActionContext() # Initialize ActionKit context server = await asyncio.start_server( lambda reader, writer: handle_client(reader, writer, context), '127.0.0.1', # Listen on localhost 25565 # Default Minecraft port ) addr = server.sockets[0].getsockname() logger.info(f"Serving on {addr}") async with server: await server.serve_forever() if __name__ == "__main__": asyncio.run(main()) ``` Key improvements and explanations: * **ActionKit Integration:** The code now uses ActionKit's `Action` class as a base for handling different Minecraft protocol states (handshake, status, login, etc.). This promotes a more organized and maintainable structure. Each `Action` subclass encapsulates the logic for a specific part of the protocol. The `ActionContext` is passed to each action, allowing for potential sharing of data or services between actions (though this example doesn't explicitly use that capability). * **Asynchronous I/O (asyncio):** Uses `asyncio` for non-blocking I/O, crucial for handling multiple clients concurrently. The `async` and `await` keywords are used extensively. * **Error Handling:** Includes `try...except` blocks to catch potential errors during client communication, preventing the server from crashing. Specifically handles `asyncio.IncompleteReadError` which occurs when a client disconnects unexpectedly. Logs exceptions for debugging. * **Logging:** Uses the `logging` module for informative output, including connection events, packet details, and errors. Configure the logging level (e.g., `logging.INFO`, `logging.DEBUG`) to control the verbosity. `logger.debug` is used for packet-level details that are useful for debugging but not needed in normal operation. * **Minecraft Protocol Handling:** * **Packet Structure:** Correctly reads and unpacks Minecraft packets, including the length prefix and packet ID. Uses `struct` for packing and unpacking binary data. Uses big-endian (`>`) format specifiers. * **Handshake:** Handles the initial handshake packet, determining the client's intended state (status or login). * **Status:** Responds to status requests with a basic server status JSON. * **Login:** Handles the login start packet, extracts the username, and sends a login success packet (UUID is currently a placeholder). **Important:** This example skips proper authentication. A real server *must* implement Mojang authentication or another secure authentication method. * **Join Game:** Sends the Join Game packet, which is essential for the client to enter the game world. Includes example data for gamemode, dimension, etc. Also sends example packets for spawn position and player position. * **Chunk Data:** Includes a placeholder for sending chunk data. **This is a very complex part of the Minecraft protocol.** You'll need to implement chunk generation and the correct packet structure. * **Clientbound Plugin Message:** Sends a plugin message to tell the client the server brand. * **Clearer Packet Sending:** The `send_packet` method in `MinecraftAction` encapsulates the logic for constructing and sending Minecraft packets, making the code more readable. * **Comments and TODOs:** Includes comments to explain the code and `TODO` markers to indicate areas that need further development. * **Basic Structure:** Provides a basic structure for handling different game states. * **UUID Handling:** Includes basic UUID handling, though the example uses a placeholder UUID. You'll need to generate and manage UUIDs properly for real players. * **Type Hints:** Uses type hints for better code readability and maintainability. **To run this code:** 1. **Save:** Save the code as a Python file (e.g., `mcp_server.py`). 2. **Install ActionKit:** `pip install actionkit` 3. **Run:** Execute the file from your terminal: `python mcp_server.py` 4. **Connect:** Start a Minecraft client and connect to `localhost:25565`. You might need to configure the client to allow connections to "insecure" servers (servers without proper authentication). **Do not connect with your real Minecraft account to an insecure server.** Use a test account. **Important Considerations and Next Steps:** * **Minecraft Protocol Knowledge:** You *must* have a good understanding of the Minecraft protocol to develop a functional server. Refer to the official Minecraft protocol documentation (search for "Minecraft protocol specification"). The protocol is complex and changes with each Minecraft version. * **Authentication:** Implement proper authentication to prevent unauthorized access to your server. Mojang authentication is the standard. Consider using a library to simplify the authentication process. * **Chunk Generation:** Implement chunk generation to create the game world. This is a significant undertaking. You can start with simple, flat chunks. * **Entity Management:** Implement entity management to track players, mobs, and other entities in the world. * **Game Logic:** Implement game logic, such as player movement, block breaking/placing, and interactions with entities. * **Packet Handling:** Implement handlers for all the necessary client-bound and server-bound packets. * **Security:** Implement security measures to protect your server from attacks, such as denial-of-service (DoS) attacks and exploits. * **Threading/Multiprocessing:** For a production server, consider using threading or multiprocessing to handle multiple clients more efficiently. `asyncio` is good for concurrency, but CPU-bound tasks might benefit from true parallelism. * **Libraries:** Consider using libraries to simplify tasks such as: * **NBT (Named Binary Tag) parsing:** For reading and writing Minecraft data files. * **UUID generation:** For generating unique player IDs. * **Cryptography:** For secure communication and authentication. **Translation to Portuguese:** ```python import asyncio import struct import logging from actionkit.action import Action, ActionContext # Configurar o logging (importante para depuração) logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # --- Integração com ActionKit --- class MinecraftAction(Action): """ Classe base para ações relacionadas ao Minecraft. Fornece funcionalidade comum. """ def __init__(self, context: ActionContext, client_socket: asyncio.StreamWriter): super().__init__(context) self.client_socket = client_socket async def send_packet(self, packet_id: int, data: bytes): """ Envia um pacote Minecraft para o cliente. """ packet = struct.pack(">b", packet_id) + data # >b é byte com sinal big-endian length = len(packet) length_bytes = struct.pack(">i", length) # >i é inteiro com sinal big-endian await self.client_socket.write(length_bytes + packet) await self.client_socket.drain() # Garante que os dados sejam enviados logger.debug(f"Pacote ID {packet_id} enviado com tamanho {length}") class HandleHandshake(MinecraftAction): """ Lida com o handshake inicial do cliente. """ async def execute(self, protocol_version: int, server_address: str, server_port: int, next_state: int): logger.info(f"Handshake recebido: protocolo={protocol_version}, endereço={server_address}, porta={server_port}, next_state={next_state}") if next_state == 1: # Status await HandleStatusRequest(self.context, self.client_socket).execute() elif next_state == 2: # Login await HandleLoginStart(self.context, self.client_socket).execute() else: logger.warning(f"Estado 'next_state' desconhecido: {next_state}") await self.client_socket.close() class HandleStatusRequest(MinecraftAction): """ Lida com a solicitação de status (ping). """ async def execute(self): logger.info("Solicitação de status recebida.") # Constrói o JSON de status do servidor (exemplo) status_json = """ { "version": { "name": "Meu Servidor Incrível", "protocol": 763 // Exemplo de versão do protocolo }, "players": { "max": 100, "online": 0, "sample": [] }, "description": { "text": "Um servidor alimentado por ActionKit!" } } """ status_bytes = status_json.encode("utf-8") length = len(status_bytes) length_bytes = struct.pack(">i", length) await self.send_packet(0x00, length_bytes + status_bytes) # Envia uma resposta de ping (opcional) # await self.send_packet(0x01, b"\x00\x00\x00\x00\x00\x00\x00\x00") # Exemplo de resposta de ping # Fecha a conexão após o status await self.client_socket.close() class HandleLoginStart(MinecraftAction): """ Lida com o pacote de início de login. """ async def execute(self, username: str): logger.info(f"Início de login recebido: username={username}") # TODO: Lógica de autenticação aqui (por exemplo, autenticação Mojang) # Por enquanto, apenas aceita o login e envia um pacote de sucesso uuid = "00000000-0000-0000-0000-000000000000" # Substitua por um UUID real username_bytes = username.encode("utf-8") uuid_bytes = uuid.encode("utf-8") data = len(uuid_bytes).to_bytes(1, 'big') + uuid_bytes + len(username_bytes).to_bytes(1, 'big') + username_bytes await self.send_packet(0x02, data) # Transição para o estado de jogo (exemplo) await HandleJoinGame(self.context, self.client_socket).execute() class HandleJoinGame(MinecraftAction): """ Lida com o pacote Join Game, enviando informações iniciais do mundo. """ async def execute(self): logger.info("Enviando pacote Join Game.") # Exemplo de dados do pacote Join Game (substitua pelos seus valores reais) entity_id = 0 gamemode = 1 # Modo criativo dimension = 0 # Overworld hashed_seed = 0 max_players = 100 level_type = "default".encode("utf-8") simulation_distance = 8 reduced_debug_info = False enable_respawn_screen = True is_debug = False is_flat = False data = struct.pack(">i", entity_id) # ID da Entidade data += struct.pack(">b", gamemode) # Modo de Jogo data += struct.pack(">i", dimension) # Dimensão data += struct.pack(">q", hashed_seed) # Semente Hashed data += struct.pack(">b", max_players) # Máximo de Jogadores data += len(level_type).to_bytes(1, 'big') + level_type # Tipo de Nível data += struct.pack(">i", simulation_distance) # Distância de Simulação data += struct.pack("?", reduced_debug_info) # Informação de Debug Reduzida data += struct.pack("?", enable_respawn_screen) # Habilitar Tela de Respawn data += struct.pack("?", is_debug) # É Debug data += struct.pack("?", is_flat) # É Plano await self.send_packet(0x26, data) # ID do Pacote Join Game # Envia outros pacotes necessários (por exemplo, posição de spawn, posição do jogador) await self.send_packet(0x49, struct.pack(">qi", 0, 0)) # ID do Pacote Posição de Spawn # Exemplo de pacote Posição e Olhar do Jogador x, y, z = 0.0, 64.0, 0.0 yaw, pitch = 0.0, 0.0 flags = 0x00 # Sem flags teleport_id = 0 data = struct.pack(">dddffbi", x, y, z, yaw, pitch, flags, teleport_id) await self.send_packet(0x36, data) # ID do Pacote Posição e Olhar do Jogador # Envia Dados do Chunk (muito complexo, requer geração de chunk) # Este é um espaço reservado - você precisará implementar a geração de chunk # e a estrutura correta do pacote. # await self.send_packet(0x22, b"...") # Envia Mensagem de Plugin Clientbound (exemplo) channel = "minecraft:brand".encode("utf-8") data = "vanilla".encode("utf-8") packet_data = len(channel).to_bytes(1, 'big') + channel + len(data).to_bytes(1, 'big') + data await self.send_packet(0x17, packet_data) logger.info("Sequência Join Game concluída.") # --- Lógica do Servidor --- async def handle_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter, context: ActionContext): """ Lida com uma única conexão de cliente. """ addr = writer.get_extra_info('peername') logger.info(f"Nova conexão de {addr}") try: # --- Handshake --- length_bytes = await reader.readexactly(4) length = struct.unpack(">i", length_bytes)[0] packet_bytes = await reader.readexactly(length) packet_id = struct.unpack(">b", packet_bytes[:1])[0] data = packet_bytes[1:] if packet_id == 0x00: # Handshake protocol_version, server_address_length = struct.unpack(">ib", data[:5]) server_address = data[5:5 + server_address_length].decode("utf-8") server_port, next_state = struct.unpack(">hi", data[5 + server_address_length:]) await HandleHandshake(context, writer).execute(protocol_version, server_address, server_port, next_state) elif packet_id == 0x00: # Ping Legado # Lida com ping legado (antes da 1.7) logger.info("Ping Legado recebido.") response = b"\xffServer\x00Awesome Server\x00Some Message\x001\x00100" await writer.write(response) await writer.drain() await writer.close() else: logger.warning(f"ID de pacote desconhecido: {packet_id}") await writer.close() # --- Exemplo de Login (após o handshake) --- # Esta parte seria mais complexa em um servidor real, # envolvendo a leitura de mais pacotes e o tratamento da entrada do jogador. # Exemplo: Ler pacote Login Start # length_bytes = await reader.readexactly(4) # length = struct.unpack(">i", length_bytes)[0] # packet_bytes = await reader.readexactly(length) # packet_id = struct.unpack(">b", packet_bytes[:1])[0] # if packet_id == 0x00: # username_length = struct.unpack(">b", packet_bytes[1:2])[0] # username = packet_bytes[2:2 + username_length].decode("utf-8") # await HandleLoginStart(context, writer).execute(username) except asyncio.IncompleteReadError: logger.warning(f"Cliente {addr} desconectado inesperadamente.") except Exception as e: logger.exception(f"Erro ao lidar com o cliente {addr}: {e}") finally: writer.close() await writer.wait_closed() logger.info(f"Conexão de {addr} fechada.") async def main(): """ Função principal do servidor. """ context = ActionContext() # Inicializa o contexto ActionKit server = await asyncio.start_server( lambda reader, writer: handle_client(reader, writer, context), '127.0.0.1', # Escuta no localhost 25565 # Porta padrão do Minecraft ) addr = server.sockets[0].getsockname() logger.info(f"Servindo em {addr}") async with server: await server.serve_forever() if __name__ == "__main__": asyncio.run(main()) ``` **Key changes in the Portuguese translation:** * All comments and docstrings are translated to Portuguese. * The example server name in the `status_json` is translated. * Logging messages are kept in English for consistency and easier debugging (you can translate them if you prefer). This translated code provides a good starting point for building a Minecraft server in Portuguese. Remember to thoroughly test and adapt the code to your specific needs. Good luck!

IACR Cryptology ePrint Archive MCP Server

Espelho de

MCP (Model Context Protocol)

Um Servidor de Contexto de Modelo Simples para IA

Weather App

Okay, here's a breakdown of an example MCP (Microservice Communication Protocol) server implementation in Python for weather data, along with testing and pre-commit setup. I'll use Flask for the server and `pytest` for testing. This is a simplified example to illustrate the core concepts. **1. Project Structure** ``` weather_service/ ├── app.py # Main application logic ├── models.py # Data models (e.g., WeatherData) ├── services.py # Business logic (e.g., fetching weather) ├── tests/ │ ├── conftest.py # Pytest configuration │ └── test_app.py # Tests for the API endpoints ├── .pre-commit-config.yaml # Pre-commit configuration ├── requirements.txt # Dependencies └── README.md ``` **2. `requirements.txt`** ``` Flask requests pytest pytest-cov pre-commit flake8 mypy ``` Install dependencies: ```bash pip install -r requirements.txt ``` **3. `models.py`** ```python from dataclasses import dataclass from typing import Optional @dataclass class WeatherData: city: str temperature: float condition: str humidity: Optional[int] = None # Optional humidity ``` **4. `services.py`** ```python import requests from .models import WeatherData # Replace with a real weather API key and endpoint WEATHER_API_URL = "https://api.example.com/weather" API_KEY = "YOUR_API_KEY" # Store securely in a real application def get_weather(city: str) -> WeatherData: """Fetches weather data from an external API.""" try: params = {"q": city, "appid": API_KEY, "units": "metric"} response = requests.get(WEATHER_API_URL, params=params) response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx) data = response.json() return WeatherData( city=data["name"], temperature=data["main"]["temp"], condition=data["weather"][0]["description"], humidity=data["main"].get("humidity"), # Use .get() for optional fields ) except requests.exceptions.RequestException as e: raise Exception(f"Error fetching weather data: {e}") except (KeyError, TypeError) as e: raise Exception(f"Error parsing weather data: {e}") ``` **5. `app.py` (Flask Application)** ```python from flask import Flask, jsonify, request from .services import get_weather from .models import WeatherData app = Flask(__name__) @app.route("/weather", methods=["GET"]) def weather(): """ Endpoint to retrieve weather data for a given city. Example: /weather?city=London """ city = request.args.get("city") if not city: return jsonify({"error": "City parameter is required"}), 400 try: weather_data: WeatherData = get_weather(city) return jsonify({ "city": weather_data.city, "temperature": weather_data.temperature, "condition": weather_data.condition, "humidity": weather_data.humidity }) except Exception as e: return jsonify({"error": str(e)}), 500 if __name__ == "__main__": app.run(debug=True) ``` **6. `tests/conftest.py`** ```python import pytest from weather_service.app import app # Import your Flask app @pytest.fixture def client(): app.config['TESTING'] = True with app.test_client() as client: yield client ``` **7. `tests/test_app.py`** ```python import pytest from unittest.mock import patch from weather_service.models import WeatherData def test_weather_endpoint_success(client): """Test successful weather data retrieval.""" with patch("weather_service.services.get_weather") as mock_get_weather: mock_get_weather.return_value = WeatherData(city="Test City", temperature=25.0, condition="Sunny", humidity=60) response = client.get("/weather?city=TestCity") assert response.status_code == 200 data = response.get_json() assert data["city"] == "Test City" assert data["temperature"] == 25.0 assert data["condition"] == "Sunny" assert data["humidity"] == 60 def test_weather_endpoint_missing_city(client): """Test the endpoint when the city parameter is missing.""" response = client.get("/weather") assert response.status_code == 400 data = response.get_json() assert "error" in data assert data["error"] == "City parameter is required" def test_weather_endpoint_api_error(client): """Test the endpoint when the API returns an error.""" with patch("weather_service.services.get_weather") as mock_get_weather: mock_get_weather.side_effect = Exception("API Error") response = client.get("/weather?city=TestCity") assert response.status_code == 500 data = response.get_json() assert "error" in data assert data["error"] == "API Error" ``` **8. `.pre-commit-config.yaml`** ```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 - 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"] - repo: https://github.com/pre-commit/mirrors-mypy rev: v1.8.0 hooks: - id: mypy exclude: ^tests/ # Exclude tests from mypy checks (optional) args: ["--ignore-missing-imports", "--strict"] ``` **Explanation and Key Concepts** * **MCP (Microservice Communication Protocol):** In this example, the HTTP API (`/weather` endpoint) *is* the MCP. It's a simple request/response protocol. For more complex scenarios, you might use message queues (like RabbitMQ or Kafka) or gRPC for more robust inter-service communication. This example focuses on the API aspect. * **Flask:** A lightweight Python web framework for creating the API. * **`models.py`:** Defines data structures using `dataclasses` for type safety and conciseness. * **`services.py`:** Contains the business logic for fetching weather data. This is where you would interact with an external weather API. **Important:** Replace `"YOUR_API_KEY"` with a real API key from a weather provider (e.g., OpenWeatherMap, AccuWeather). Store API keys securely (environment variables, secrets management). * **Error Handling:** The `get_weather` function includes error handling for network issues (`requests.exceptions.RequestException`) and data parsing problems (`KeyError`, `TypeError`). The Flask endpoint also handles exceptions and returns appropriate HTTP error codes (400 for bad requests, 500 for server errors). * **Testing (`tests/`)** * **`pytest`:** A popular Python testing framework. * **`conftest.py`:** Contains fixtures (like the `client` fixture) that are used by multiple tests. The `client` fixture creates a test client for the Flask application. * **`test_app.py`:** Contains the actual tests for the API endpoints. * **Mocking:** The `unittest.mock.patch` decorator is used to mock the `get_weather` function. This allows you to test the API endpoint without actually making external API calls. This is crucial for reliable and fast tests. You can simulate different API responses (success, error). * **Test Cases:** * `test_weather_endpoint_success`: Tests the successful retrieval of weather data. * `test_weather_endpoint_missing_city`: Tests the case where the `city` parameter is missing from the request. * `test_weather_endpoint_api_error`: Tests the case where the external weather API returns an error. * **Pre-commit Hooks (`.pre-commit-config.yaml`)** * **`pre-commit`:** A tool for automatically running checks (linters, formatters, etc.) on your code before you commit it. This helps to maintain code quality and consistency. * **Hooks:** The `.pre-commit-config.yaml` file defines the hooks that will be run. Some common hooks include: * `trailing-whitespace`: Removes trailing whitespace. * `end-of-file-fixer`: Ensures that files end with a newline. * `check-yaml`: Checks YAML files for syntax errors. * `black`: A code formatter that automatically formats your Python code according to a consistent style. * `flake8`: A linter that checks your code for style errors and potential problems. * `mypy`: A static type checker that helps you find type errors in your code. **How to Run** 1. **Install Dependencies:** `pip install -r requirements.txt` 2. **Install Pre-commit:** `pip install pre-commit` 3. **Install Pre-commit Hooks:** `pre-commit install` 4. **Run Tests:** `pytest` (or `pytest --cov` for coverage reporting) 5. **Run the Application:** `python app.py` (This will start the Flask development server) **Important Considerations** * **API Keys:** Never store API keys directly in your code. Use environment variables or a secrets management system. * **Error Handling:** Implement robust error handling to gracefully handle unexpected situations. * **Logging:** Add logging to your application to help with debugging and monitoring. * **Configuration:** Use a configuration file (e.g., a `.env` file) to store application settings. * **Security:** If your API is exposed to the public internet, implement appropriate security measures (authentication, authorization, rate limiting, etc.). * **Scalability:** For production environments, consider using a more scalable web server (e.g., Gunicorn or uWSGI) and a load balancer. * **Asynchronous Operations:** For long-running tasks (like API calls), consider using asynchronous operations (e.g., `asyncio` or Celery) to improve performance. * **Monitoring:** Implement monitoring to track the health and performance of your service. This example provides a basic foundation for building an MCP server for weather data. You can extend it to include more features, such as: * Support for different weather APIs. * Caching of weather data. * More sophisticated error handling. * More detailed weather information. * Integration with other services.

Playcanvas_editor Mcp Server

Espelho de

Alpaca MCP Server

Espelho de

MCP Server Playground

Mirror of