diff --git a/src/oci-limits-mcp-server/.gitignore b/src/oci-limits-mcp-server/.gitignore new file mode 100644 index 0000000..9dcdf3e --- /dev/null +++ b/src/oci-limits-mcp-server/.gitignore @@ -0,0 +1,68 @@ +# OS / VCS +.DS_Store +Thumbs.db +.git/ +.git-* +*.orig + +# Python bytecode +__pycache__/ +*.py[cod] +*$py.class + +# Virtual env +.venv/ +venv/ +env/ +ENV/ +.conda/ +.mamba/ + +# Packaging / build +build/ +dist/ +*.egg-info/ +.eggs/ +pip-wheel-metadata/ +wheels/ +*.egg +MANIFEST + +# Test / coverage +.pytest_cache/ +.coverage +.coverage.* +htmlcov/ +.cache/ +.nox/ +.tox/ + +# Type checking / linters +.mypy_cache/ +.ruff_cache/ +.pyre/ + +# Logs / temp +*.log +logs/ +*.pid +*.tmp +*.swp + +# Notebooks +.ipynb_checkpoints/ + +# IDE +.vscode/ +.idea/ +*.code-workspace + +# Secrets / local env +.env +.env.* +*.pem +*.key +*.crt + +# Docker (repo 内で不要なローカル生成物) +docker-compose.override.yml diff --git a/src/oci-limits-mcp-server/Dockerfile b/src/oci-limits-mcp-server/Dockerfile new file mode 100644 index 0000000..3e62fd1 --- /dev/null +++ b/src/oci-limits-mcp-server/Dockerfile @@ -0,0 +1,35 @@ +FROM python:3.11-slim + +# Install system dependencies +RUN apt-get update && apt-get install -y \ + gcc \ + && rm -rf /var/lib/apt/lists/* + +# Set working directory +WORKDIR /app + +# Install uv +RUN pip install uv + +# Copy requirements and install dependencies +COPY requirements.txt pyproject.toml ./ +RUN uv pip install --system -r requirements.txt + +# Copy source code +COPY oci_limits_mcp_server.py ./ + +# Create non-root user for security +RUN useradd --create-home --shell /bin/bash oci +RUN chown -R oci:oci /app +USER oci + +# Set environment variables +ENV PYTHONUNBUFFERED=1 +ENV PYTHONPATH=/app + +# Health check +HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \ + CMD python -c "import oci_limits_mcp_server; print('OK')" + +# Run the server +CMD ["python", "oci_limits_mcp_server.py"] diff --git a/src/oci-limits-mcp-server/README.md b/src/oci-limits-mcp-server/README.md new file mode 100644 index 0000000..3cecffe --- /dev/null +++ b/src/oci-limits-mcp-server/README.md @@ -0,0 +1,341 @@ +# OCI Service Limits and Quota Policies MCP Server + +A Python-based Model Context Protocol (MCP) server that provides **read-only access** to Oracle Cloud Infrastructure (OCI) service limits and quota policies. Query resource limits, quotas, and availability information across OCI services and regions using natural language. + +## 🚀 Quick Start + +### Prerequisites +- Python 3.11+ +- Valid OCI configuration +- uv package manager (recommended) or Docker/Podman + +### Install and Run +```bash +# Clone and navigate +cd /path/to/oci_limits_mcp_server + +# Install dependencies +uv sync + +# Run the server +uv run python oci_limits_mcp_server.py +``` + +## 🛠️ Available MCP Tools + +The server provides 10 MCP tools for querying OCI limits and quotas: + +### Service Limits Tools +1. **`list_supported_services()`** - List all OCI services with configurable limits +2. **`list_service_limits(service_name, compartment_name?, availability_domain?, limit_name?)`** - Get all limits for a service +3. **`get_service_limit(service_name, limit_name, compartment_name?, availability_domain?)`** - Get specific limit details +4. **`check_service_limits_by_region(service_name, region, compartment_name?)`** - Check limits in a specific region + +### Quota Policy Tools +5. **`list_quota_policies(compartment_name?)`** - List all quota policies in a compartment +6. **`get_quota_policy(quota_name, compartment_name?)`** - Get detailed quota policy information + +### Resource Discovery Tools +7. **`list_all_compartments()`** - List all compartments in your tenancy +8. **`list_availability_domains(compartment_name?)`** - List availability domains +9. **`get_resource_availability(service_name, resource_type, compartment_name?, availability_domain?)`** - Check resource availability + +### Utility Tools +10. **`ping()`** - Health check endpoint + +## 📋 Setup Methods + +## Method 1: Using uv (Recommended) + +### Install uv +```bash +# Install uv if you haven't already +curl -LsSf https://astral.sh/uv/install.sh | sh +# or +pip install uv +``` + +### Setup and Run +```bash +# Navigate to the server directory +cd /Users/your-username/path/to/oracle_mcp/src/oci_limits_mcp_server + +# Install dependencies +uv sync --dev + +# Test the server +uv run python oci_limits_mcp_server.py --help + +# Run with specific OCI profile +PROFILE_NAME=YOUR_PROFILE uv run python oci_limits_mcp_server.py + +# Run tests +uv run python -m pytest test_oci_limits_mcp_server.py -v +``` + +## Method 2: Using Docker/Podman + +### Build Container +```bash +# Navigate to server directory +cd /Users/your-username/path/to/oracle_mcp/src/oci_limits_mcp_server + +# Build with Docker +docker build -t oci-limits-mcp:latest . + +# Or build with Podman +podman build -t oci-limits-mcp:latest . +``` + +### Run Container +```bash +# Run without OCI config (will show warnings but work for testing) +podman run --rm -it oci-limits-mcp:latest + +# Run with OCI config mounted +podman run --rm -it \ + -v ~/.oci:/home/oci/.oci:ro \ + -e PROFILE_NAME=YOUR_PROFILE \ + oci-limits-mcp:latest +``` + +## 🔧 OCI Configuration + +### 1. Create OCI Config Directory +```bash +mkdir -p ~/.oci +chmod 700 ~/.oci +``` + +### 2. Create OCI Config File (`~/.oci/config`) +```ini +[DEFAULT] +user=ocid1.user.oc1..your_user_ocid +fingerprint=aa:bb:cc:dd:ee:ff:11:22:33:44:55:66:77:88:99:00:aa:bb:cc:dd +key_file=~/.oci/oci_api_key.pem +tenancy=ocid1.tenancy.oc1..your_tenancy_ocid +region=us-ashburn-1 + +[DEV] +user=ocid1.user.oc1..your_dev_user_ocid +fingerprint=bb:cc:dd:ee:ff:11:22:33:44:55:66:77:88:99:00:aa:bb:cc:dd:ee +key_file=~/.oci/oci_api_key_dev.pem +tenancy=ocid1.tenancy.oc1..your_dev_tenancy_ocid +region=us-phoenix-1 +``` + +### 3. Set Key File Permissions +```bash +chmod 600 ~/.oci/oci_api_key.pem +``` + +### 4. Required OCI IAM Permissions +``` +Allow group to read limits in tenancy +Allow group to read quotas in tenancy +Allow group to read compartments in tenancy +Allow group to read availability-domains in tenancy +``` + +## 🔌 Integration Methods + +## Integration 1: VS Code with GitHub Copilot + +### Setup Steps: +1. **Open VS Code Settings JSON:** + - Press `Cmd+Shift+P` → "Preferences: Open User Settings (JSON)" + +2. **Add MCP Configuration:** +```json +{ + "github.copilot.chat.mcp.servers": { + "oci-limits": { + "command": "uv", + "args": [ + "run", + "--project", + "/Users/your-username/path/to/oracle_mcp/src/oci_limits_mcp_server", + "python", + "oci_limits_mcp_server.py" + ], + "env": { + "PROFILE_NAME": "DEFAULT" + } + } + } +} +``` + +3. **Restart VS Code** + +4. **Test with GitHub Copilot:** + - Open Copilot Chat (`Cmd+Shift+P` → "GitHub Copilot: Open Chat") + - Try: "What are my OCI compute limits in us-ashburn-1?" + +## Integration 2: Claude Desktop + +### Setup Steps: +1. **Create/Edit Claude Config:** +```bash +# Create config file +mkdir -p ~/Library/Application\ Support/Claude +``` + +2. **Add Configuration (`~/Library/Application Support/Claude/claude_desktop_config.json`):** +```json +{ + "mcpServers": { + "oci-limits": { + "command": "uv", + "args": [ + "run", + "--project", + "/Users/your-username/path/to/oracle_mcp/src/oci_limits_mcp_server", + "python", + "oci_limits_mcp_server.py" + ], + "env": { + "PROFILE_NAME": "DEFAULT" + } + } + } +} +``` + +3. **Restart Claude Desktop** + +4. **Test in Claude:** + - Ask: "Show me my OCI service limits" + - Ask: "What quota policies do I have?" + +## Integration 3: Cline Extension + +### Setup Steps: +1. **Install Cline Extension:** +```bash +code --install-extension saoudrizwan.claude-dev +``` + +2. **The MCP config should already exist at:** +``` +/Users/your-username/path/to/oracle_mcp/.vscode/mcp.json +``` + +3. **Test with Cline:** + - Open Cline panel in VS Code + - Ask: "Can you check my OCI compute limits?" + - Ask: "What are my quota policies in the Dev compartment?" + +## 📝 Testing + +### Run the Test Suite +```bash +cd /Users/your-username/path/to/oracle_mcp/src/oci_limits_mcp_server + +# Run all tests +uv run python -m pytest test_oci_limits_mcp_server.py -v + +# Run with coverage +uv run python -m pytest test_oci_limits_mcp_server.py --cov=oci_limits_mcp_server +``` + +### Test Different Profiles +```bash +# Test with specific profile +PROFILE_NAME=DEV uv run python oci_limits_mcp_server.py + +# Test import +uv run python -c "import oci_limits_mcp_server; print('Import successful')" +``` + +## 💬 Example Queries + +### Service Limits Queries +- "What compute limits do I have in us-ashburn-1?" +- "Show me all block storage limits" +- "What's the maximum number of VCNs I can create?" +- "List all services that have configurable limits" + +### Quota Policy Queries +- "Show me all quota policies in my tenancy" +- "What are the quota statements for my compute quota?" +- "Am I close to any quota limits in my Dev compartment?" + +### Resource Availability Queries +- "What compute shapes are available in AD-1?" +- "Show me availability domains in us-phoenix-1" +- "Check resource availability for block storage" + +## 📊 Example Responses + +### Service Limits Response +```json +{ + "service": "compute", + "total_limits": 15, + "limits": [ + { + "service_name": "compute", + "name": "standard-a1-core-count", + "description": "Number of A1 cores", + "value": 1000, + "scope_type": "AD", + "availability_domain": "Uocm:US-ASHBURN-AD-1", + "compartment_name": "root" + } + ] +} +``` + +### Quota Policy Response +```json +{ + "id": "ocid1.quota.oc1..example", + "name": "compute-quota", + "description": "Quota for compute resources", + "compartment_name": "Dev", + "lifecycle_state": "ACTIVE", + "statements": [ + "set compute quota standard-a1-core-count to 500 in compartment Dev" + ] +} +``` + +## 🔧 Troubleshooting + +### Common Issues + +#### "MCP Server Not Connecting" +- Verify file paths in configuration are absolute and correct +- Ensure `uv` is installed: `uv --version` +- Check OCI configuration: `uv run python -c "import oci; print('OCI SDK works')"` + +#### "Profile Not Found" +- Check profile exists: `cat ~/.oci/config` +- Verify profile name spelling (case-sensitive) +- Test with: `PROFILE_NAME=YOUR_PROFILE uv run python oci_limits_mcp_server.py` + +#### "Permission Errors" +- Check key file permissions: `ls -la ~/.oci/` +- Verify IAM permissions in OCI Console +- Test OCI CLI: `oci iam user get --user-id ` + +### Debug Mode +```bash +# Enable debug logging +export MCP_DEBUG=1 +export OCI_PYTHON_SDK_DEBUG=1 +PROFILE_NAME=YOUR_PROFILE uv run python oci_limits_mcp_server.py +``` + +## 🔒 Security Notes + +- **Read-only access**: Only performs read operations on OCI APIs +- **No credential storage**: Uses standard OCI SDK configuration +- **Stateless**: No data persistence or storage +- **Principle of least privilege**: Only requires read permissions + +## 📜 License + +Copyright (c) 2025, Oracle and/or its affiliates. +Licensed under the Universal Permissive License v1.0 as shown at http://oss.oracle.com/licenses/upl. diff --git a/src/oci-limits-mcp-server/oci_limits_mcp_server.py b/src/oci-limits-mcp-server/oci_limits_mcp_server.py new file mode 100644 index 0000000..57cabe2 --- /dev/null +++ b/src/oci-limits-mcp-server/oci_limits_mcp_server.py @@ -0,0 +1,552 @@ +""" +Copyright (c) 2025, Oracle and/or its affiliates. +Licensed under the Universal Permissive License v1.0 as shown at http://oss.oracle.com/licenses/upl. + +OCI Service Limits and Quota Policies MCP Server +- Provides read-only access to OCI service limits and quota policies +- Supports querying limits by service, region, and compartment +- Allows checking current usage against quota limits +- Returns structured JSON for easy consumption by LLMs +""" + +from __future__ import annotations + +import json +import os +from typing import Any, Dict, List, Optional + +import oci +from fastmcp import FastMCP + +# Initialize FastMCP +mcp = FastMCP("oci-limits") + +# Configuration +profile_name = os.getenv("PROFILE_NAME", "DEFAULT") + +try: + config = oci.config.from_file(profile_name=profile_name) + identity_client = oci.identity.IdentityClient(config) + limits_client = oci.limits.LimitsClient(config) + usage_api_client = oci.usage_api.UsageapiClient(config) + monitoring_client = oci.monitoring.MonitoringClient(config) + resource_search_client = oci.resource_search.ResourceSearchClient(config) + tenancy_id = os.getenv("TENANCY_ID_OVERRIDE", config['tenancy']) +except Exception as e: + print(f"Warning: Failed to initialize OCI clients: {e}") + config = None + identity_client = None + limits_client = None + usage_api_client = None + monitoring_client = None + resource_search_client = None + tenancy_id = None + + +def get_compartment_by_name(compartment_name: str): + """Internal function to get compartment by name""" + if not identity_client: + return None + + try: + compartments = identity_client.list_compartments( + compartment_id=tenancy_id, + compartment_id_in_subtree=True, + access_level="ACCESSIBLE", + lifecycle_state="ACTIVE" + ) + # Add root compartment (tenancy) + compartments.data.append(identity_client.get_compartment(compartment_id=tenancy_id).data) + + # Search for the compartment by name + for compartment in compartments.data: + if compartment.name.lower() == compartment_name.lower(): + return compartment + return None + except Exception as e: + print(f"Error getting compartment: {e}") + return None + + +def format_error_response(error_message: str, context: Optional[Dict[str, Any]] = None) -> str: + """Format error responses consistently""" + response = {"error": error_message} + if context: + response.update(context) + return json.dumps(response, indent=2) + + +@mcp.tool() +def list_all_compartments() -> str: + """List all compartments in the tenancy""" + if not identity_client: + return format_error_response("OCI client not initialized") + + try: + compartments = identity_client.list_compartments( + compartment_id=tenancy_id, + compartment_id_in_subtree=True, + access_level="ACCESSIBLE", + lifecycle_state="ACTIVE" + ).data + + # Add root compartment (tenancy) + root_compartment = identity_client.get_compartment(compartment_id=tenancy_id).data + compartments.append(root_compartment) + + # Format for better readability + result = [] + for compartment in compartments: + result.append({ + "id": compartment.id, + "name": compartment.name, + "description": compartment.description, + "lifecycle_state": compartment.lifecycle_state, + "time_created": str(compartment.time_created) + }) + + return json.dumps(result, indent=2) + except Exception as e: + return format_error_response(f"Failed to list compartments: {str(e)}") + + +@mcp.tool() +def list_service_limits(service_name: str, compartment_name: Optional[str] = None, + availability_domain: Optional[str] = None, limit_name: Optional[str] = None) -> str: + """ + List service limits for a specific service in OCI + + Args: + service_name: Name of the OCI service (e.g., "compute", "block-storage", "database") + compartment_name: Optional compartment name to scope the limits + availability_domain: Optional availability domain filter + limit_name: Optional specific limit name to filter by + """ + if not limits_client: + return format_error_response("OCI limits client not initialized") + + try: + compartment_id = tenancy_id + if compartment_name: + compartment = get_compartment_by_name(compartment_name) + if not compartment: + return format_error_response(f"Compartment '{compartment_name}' not found") + compartment_id = compartment.id + + # List limits for the service + kwargs = { + "service_name": service_name, + "compartment_id": compartment_id + } + + if availability_domain: + kwargs["availability_domain"] = availability_domain + if limit_name: + kwargs["name"] = limit_name + + limits_response = limits_client.list_limit_values(**kwargs) + + result = [] + for limit in limits_response.data: + limit_info = { + "service_name": service_name, + "name": limit.name, + "description": getattr(limit, 'description', None), + "value": limit.value, + "scope_type": limit.scope_type, + "availability_domain": limit.availability_domain, + "compartment_id": compartment_id, + "compartment_name": compartment_name or "root" + } + result.append(limit_info) + + return json.dumps({ + "service": service_name, + "total_limits": len(result), + "limits": result + }, indent=2) + + except Exception as e: + return format_error_response(f"Failed to list service limits: {str(e)}", { + "service_name": service_name, + "compartment_name": compartment_name + }) + + +@mcp.tool() +def get_service_limit(service_name: str, limit_name: str, compartment_name: Optional[str] = None, + availability_domain: Optional[str] = None) -> str: + """ + Get details for a specific service limit + + Args: + service_name: Name of the OCI service + limit_name: Specific limit name + compartment_name: Optional compartment name to scope the limit + availability_domain: Optional availability domain filter + """ + if not limits_client: + return format_error_response("OCI limits client not initialized") + + try: + compartment_id = tenancy_id + if compartment_name: + compartment = get_compartment_by_name(compartment_name) + if not compartment: + return format_error_response(f"Compartment '{compartment_name}' not found") + compartment_id = compartment.id + + # Get the specific limit + kwargs = { + "service_name": service_name, + "limit_name": limit_name, + "compartment_id": compartment_id + } + + if availability_domain: + kwargs["availability_domain"] = availability_domain + + limit_response = limits_client.get_limit_value(**kwargs) + + result = { + "service_name": service_name, + "name": limit_response.data.name, + "description": getattr(limit_response.data, 'description', None), + "value": limit_response.data.value, + "scope_type": limit_response.data.scope_type, + "availability_domain": limit_response.data.availability_domain, + "compartment_id": compartment_id, + "compartment_name": compartment_name or "root" + } + + return json.dumps(result, indent=2) + + except Exception as e: + return format_error_response(f"Failed to get service limit: {str(e)}", { + "service_name": service_name, + "limit_name": limit_name, + "compartment_name": compartment_name + }) + + +@mcp.tool() +def list_supported_services() -> str: + """List all supported services that have limits in OCI""" + if not limits_client: + return format_error_response("OCI limits client not initialized") + + try: + services_response = limits_client.list_services(compartment_id=tenancy_id) + + result = [] + for service in services_response.data: + service_info = { + "name": service.name, + "description": service.description, + } + result.append(service_info) + + return json.dumps({ + "total_services": len(result), + "services": result + }, indent=2) + + except Exception as e: + return format_error_response(f"Failed to list supported services: {str(e)}") + + +@mcp.tool() +def list_quota_policies(compartment_name: Optional[str] = None) -> str: + """ + List quota policies in a compartment + + Args: + compartment_name: Optional compartment name. If not provided, uses root compartment. + """ + if not limits_client: + return format_error_response("OCI limits client not initialized") + + try: + compartment_id = tenancy_id + if compartment_name: + compartment = get_compartment_by_name(compartment_name) + if not compartment: + return format_error_response(f"Compartment '{compartment_name}' not found") + compartment_id = compartment.id + + quotas_response = limits_client.list_quotas(compartment_id=compartment_id) + + result = [] + for quota in quotas_response.data: + quota_info = { + "id": quota.id, + "name": quota.name, + "description": quota.description, + "compartment_id": quota.compartment_id, + "lifecycle_state": quota.lifecycle_state, + "time_created": str(quota.time_created), + "freeform_tags": quota.freeform_tags, + "defined_tags": quota.defined_tags + } + result.append(quota_info) + + return json.dumps({ + "compartment_name": compartment_name or "root", + "total_quotas": len(result), + "quotas": result + }, indent=2) + + except Exception as e: + return format_error_response(f"Failed to list quota policies: {str(e)}", { + "compartment_name": compartment_name + }) + + +@mcp.tool() +def get_quota_policy(quota_name: str, compartment_name: Optional[str] = None) -> str: + """ + Get detailed information about a specific quota policy + + Args: + quota_name: Name of the quota policy + compartment_name: Optional compartment name. If not provided, uses root compartment. + """ + if not limits_client: + return format_error_response("OCI limits client not initialized") + + try: + compartment_id = tenancy_id + if compartment_name: + compartment = get_compartment_by_name(compartment_name) + if not compartment: + return format_error_response(f"Compartment '{compartment_name}' not found") + compartment_id = compartment.id + + # First, find the quota by name + quotas_response = limits_client.list_quotas(compartment_id=compartment_id) + quota_id = None + for quota in quotas_response.data: + if quota.name.lower() == quota_name.lower(): + quota_id = quota.id + break + + if not quota_id: + return format_error_response(f"Quota policy '{quota_name}' not found in compartment") + + # Get the detailed quota information + quota_response = limits_client.get_quota(quota_id=quota_id) + quota = quota_response.data + + # Get quota statements + statements = quota.statements if hasattr(quota, 'statements') and quota.statements else [] + + result = { + "id": quota.id, + "name": quota.name, + "description": quota.description, + "compartment_id": quota.compartment_id, + "compartment_name": compartment_name or "root", + "lifecycle_state": quota.lifecycle_state, + "time_created": str(quota.time_created), + "statements": statements, + "freeform_tags": quota.freeform_tags, + "defined_tags": quota.defined_tags + } + + return json.dumps(result, indent=2) + + except Exception as e: + return format_error_response(f"Failed to get quota policy: {str(e)}", { + "quota_name": quota_name, + "compartment_name": compartment_name + }) + + +@mcp.tool() +def check_service_limits_by_region(service_name: str, region: str, + compartment_name: Optional[str] = None) -> str: + """ + Check service limits for a specific service in a particular region + + Args: + service_name: Name of the OCI service + region: OCI region name (e.g., "us-ashburn-1", "us-phoenix-1") + compartment_name: Optional compartment name + """ + if not limits_client: + return format_error_response("OCI limits client not initialized") + + try: + compartment_id = tenancy_id + if compartment_name: + compartment = get_compartment_by_name(compartment_name) + if not compartment: + return format_error_response(f"Compartment '{compartment_name}' not found") + compartment_id = compartment.id + + # Create a new limits client for the specific region + region_config = config.copy() + region_config['region'] = region + region_limits_client = oci.limits.LimitsClient(region_config) + + # List limits for the service in the specific region + limits_response = region_limits_client.list_limit_values( + service_name=service_name, + compartment_id=compartment_id + ) + + result = [] + for limit in limits_response.data: + limit_info = { + "service_name": service_name, + "region": region, + "name": limit.name, + "description": getattr(limit, 'description', None), + "value": limit.value, + "scope_type": limit.scope_type, + "availability_domain": limit.availability_domain, + "compartment_id": compartment_id, + "compartment_name": compartment_name or "root" + } + result.append(limit_info) + + return json.dumps({ + "service": service_name, + "region": region, + "total_limits": len(result), + "limits": result + }, indent=2) + + except Exception as e: + return format_error_response(f"Failed to check service limits by region: {str(e)}", { + "service_name": service_name, + "region": region, + "compartment_name": compartment_name + }) + + +@mcp.tool() +def get_resource_availability(service_name: str, resource_type: str, + compartment_name: Optional[str] = None, + availability_domain: Optional[str] = None) -> str: + """ + Get resource availability information for a service in OCI + + Args: + service_name: Name of the OCI service + resource_type: Type of resource to check availability for + compartment_name: Optional compartment name + availability_domain: Optional availability domain + """ + if not limits_client: + return format_error_response("OCI limits client not initialized") + + try: + compartment_id = tenancy_id + if compartment_name: + compartment = get_compartment_by_name(compartment_name) + if not compartment: + return format_error_response(f"Compartment '{compartment_name}' not found") + compartment_id = compartment.id + + # Get resource availability + kwargs = { + "service_name": service_name, + "compartment_id": compartment_id + } + + if availability_domain: + kwargs["availability_domain"] = availability_domain + + # List all limits for the service first + limits_response = limits_client.list_limit_values(**kwargs) + + # Filter by resource type if specified + relevant_limits = [] + for limit in limits_response.data: + if not resource_type or resource_type.lower() in limit.name.lower(): + relevant_limits.append({ + "name": limit.name, + "description": getattr(limit, 'description', None), + "value": limit.value, + "scope_type": limit.scope_type, + "availability_domain": limit.availability_domain + }) + + result = { + "service_name": service_name, + "resource_type": resource_type, + "compartment_name": compartment_name or "root", + "availability_domain": availability_domain, + "total_limits": len(relevant_limits), + "limits": relevant_limits + } + + return json.dumps(result, indent=2) + + except Exception as e: + return format_error_response(f"Failed to get resource availability: {str(e)}", { + "service_name": service_name, + "resource_type": resource_type, + "compartment_name": compartment_name + }) + + +@mcp.tool() +def list_availability_domains(compartment_name: Optional[str] = None) -> str: + """ + List availability domains in a compartment or region + + Args: + compartment_name: Optional compartment name. If not provided, uses root compartment. + """ + if not identity_client: + return format_error_response("OCI identity client not initialized") + + try: + compartment_id = tenancy_id + if compartment_name: + compartment = get_compartment_by_name(compartment_name) + if not compartment: + return format_error_response(f"Compartment '{compartment_name}' not found") + compartment_id = compartment.id + + # List availability domains + ads_response = identity_client.list_availability_domains(compartment_id=compartment_id) + + result = [] + for ad in ads_response.data: + ad_info = { + "name": ad.name, + "id": ad.id, + "compartment_id": ad.compartment_id + } + result.append(ad_info) + + return json.dumps({ + "compartment_name": compartment_name or "root", + "total_availability_domains": len(result), + "availability_domains": result + }, indent=2) + + except Exception as e: + return format_error_response(f"Failed to list availability domains: {str(e)}", { + "compartment_name": compartment_name + }) + + +@mcp.tool() +def ping() -> str: + """Health check to verify the server is responsive""" + if not config: + return json.dumps({"status": "error", "message": "OCI configuration not initialized"}) + return json.dumps({"status": "ok", "message": "OCI Limits MCP Server is running"}) + + +def main() -> None: + """Start the MCP server""" + mcp.run() + + +if __name__ == "__main__": + main() diff --git a/src/oci-limits-mcp-server/pyproject.toml b/src/oci-limits-mcp-server/pyproject.toml new file mode 100644 index 0000000..c939a04 --- /dev/null +++ b/src/oci-limits-mcp-server/pyproject.toml @@ -0,0 +1,39 @@ +[project] +name = "oci-limits-mcp-server" +version = "0.1.0" +description = "OCI Service Limits and Quota Policies MCP Server" +requires-python = ">=3.11" +dependencies = [ + "fastmcp>=2.0.0", + "oci>=2.100.0", +] + +[dependency-groups] +dev = [ + "pytest>=7.0.0", + "pytest-cov>=4.0.0", + "pytest-mock>=3.10.0", +] + +[project.scripts] +oci-limits-mcp = "oci_limits_mcp_server:main" + +[tool.setuptools] +py-modules = ["oci_limits_mcp_server"] + +[tool.black] +line-length = 100 +target-version = ["py311"] + +[tool.ruff] +line-length = 100 +target-version = "py311" +fix = true + +[tool.ruff.lint] +select = ["E","F","I","UP","B"] +ignore = ["E501"] + +[build-system] +requires = ["setuptools>=61.0"] +build-backend = "setuptools.build_meta" diff --git a/src/oci-limits-mcp-server/requirements.txt b/src/oci-limits-mcp-server/requirements.txt new file mode 100644 index 0000000..efaa8e6 --- /dev/null +++ b/src/oci-limits-mcp-server/requirements.txt @@ -0,0 +1,3 @@ +# OCI Limits MCP Server Dependencies +fastmcp>=2.0.0 +oci>=2.100.0 diff --git a/src/oci-limits-mcp-server/test_oci_limits_mcp_server.py b/src/oci-limits-mcp-server/test_oci_limits_mcp_server.py new file mode 100644 index 0000000..8775a70 --- /dev/null +++ b/src/oci-limits-mcp-server/test_oci_limits_mcp_server.py @@ -0,0 +1,124 @@ +""" +Test module for OCI Limits MCP Server + +Tests the core functionality of the OCI service limits and quota policies server +""" + +import json +import unittest +from unittest.mock import Mock, patch, MagicMock +import sys +import os + +# Add the current directory to Python path for imports +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +class TestOCILimitsMCPServer(unittest.TestCase): + """Test cases for OCI Limits MCP Server""" + + def setUp(self): + """Set up test environment""" + # Mock OCI clients to avoid requiring actual OCI configuration + self.mock_config = { + 'tenancy': 'ocid1.tenancy.oc1..test', + 'region': 'us-ashburn-1' + } + + @patch('oci.config.from_file') + @patch('oci.identity.IdentityClient') + @patch('oci.limits.LimitsClient') + @patch('oci.usage_api.UsageapiClient') + @patch('oci.monitoring.MonitoringClient') + @patch('oci.resource_search.ResourceSearchClient') + def test_server_initialization(self, mock_resource_search, mock_monitoring, mock_usage_api, + mock_limits_client, mock_identity_client, mock_config): + """Test that the server initializes properly with mocked OCI clients""" + # Mock the config + mock_config.return_value = self.mock_config + + # Mock the clients + mock_identity = Mock() + mock_limits = Mock() + mock_identity_client.return_value = mock_identity + mock_limits_client.return_value = mock_limits + mock_usage_api.return_value = Mock() + mock_monitoring.return_value = Mock() + mock_resource_search.return_value = Mock() + + # Import the module after patching + import oci_limits_mcp_server + + # Verify that the module can be imported successfully + self.assertIsNotNone(oci_limits_mcp_server) + + @patch('oci.config.from_file') + @patch('oci.identity.IdentityClient') + @patch('oci.limits.LimitsClient') + @patch('oci.usage_api.UsageapiClient') + @patch('oci.monitoring.MonitoringClient') + @patch('oci.resource_search.ResourceSearchClient') + def test_format_error_response(self, mock_resource_search, mock_monitoring, mock_usage_api, + mock_limits_client, mock_identity_client, mock_config): + """Test error response formatting""" + # Mock all the clients to avoid initialization errors + mock_config.return_value = self.mock_config + mock_identity_client.return_value = Mock() + mock_limits_client.return_value = Mock() + mock_usage_api.return_value = Mock() + mock_monitoring.return_value = Mock() + mock_resource_search.return_value = Mock() + + import oci_limits_mcp_server + + # Test format_error_response function directly + error_msg = "Test error" + context = {"test": "context"} + result = oci_limits_mcp_server.format_error_response(error_msg, context) + + # Parse JSON response + parsed = json.loads(result) + self.assertEqual(parsed["error"], error_msg) + self.assertEqual(parsed["test"], "context") + + @patch('oci.config.from_file') + @patch('oci.identity.IdentityClient') + @patch('oci.limits.LimitsClient') + @patch('oci.usage_api.UsageapiClient') + @patch('oci.monitoring.MonitoringClient') + @patch('oci.resource_search.ResourceSearchClient') + def test_mcp_tools_exist(self, mock_resource_search, mock_monitoring, mock_usage_api, + mock_limits_client, mock_identity_client, mock_config): + """Test that MCP tools are properly registered""" + # Mock all the clients + mock_config.return_value = self.mock_config + mock_identity_client.return_value = Mock() + mock_limits_client.return_value = Mock() + mock_usage_api.return_value = Mock() + mock_monitoring.return_value = Mock() + mock_resource_search.return_value = Mock() + + import oci_limits_mcp_server + + # Check that the MCP server instance exists + self.assertIsNotNone(oci_limits_mcp_server.mcp) + + # Check that tools are registered + # This is a basic test to ensure the module loads correctly + # The actual FastMCP testing would require more complex setup + self.assertTrue(hasattr(oci_limits_mcp_server, 'mcp')) + + def test_client_initialization_error(self): + """Test graceful handling of OCI client initialization errors""" + with patch('oci.config.from_file', side_effect=Exception("Config not found")): + # Import should not raise an exception + import oci_limits_mcp_server + + # The module should still be importable + self.assertIsNotNone(oci_limits_mcp_server) + + # Check that error handling works + self.assertIsNone(oci_limits_mcp_server.identity_client) + self.assertIsNone(oci_limits_mcp_server.limits_client) + +if __name__ == '__main__': + unittest.main()