diff --git a/src/oci-limits-mcp-server/LICENSE.txt b/src/oci-limits-mcp-server/LICENSE.txt new file mode 100644 index 0000000..8dc7c07 --- /dev/null +++ b/src/oci-limits-mcp-server/LICENSE.txt @@ -0,0 +1,35 @@ +Copyright (c) 2025 Oracle and/or its affiliates. + +The Universal Permissive License (UPL), Version 1.0 + +Subject to the condition set forth below, permission is hereby granted to any +person obtaining a copy of this software, associated documentation and/or data +(collectively the "Software"), free of charge and under any and all copyright +rights in the Software, and any and all patent rights owned or freely +licensable by each licensor hereunder covering either (i) the unmodified +Software as contributed to or provided by such licensor, or (ii) the Larger +Works (as defined below), to deal in both + +(a) the Software, and +(b) any piece of software and/or hardware listed in the lrgrwrks.txt file if +one is included with the Software (each a "Larger Work" to which the Software +is contributed by such licensors), + +without restriction, including without limitation the rights to copy, create +derivative works of, display, perform, and distribute the Software and make, +use, sell, offer for sale, import, export, have made, and have sold the +Software and the Larger Work(s), and to sublicense the foregoing rights on +either these or other terms. + +This license is subject to the following condition: +The above copyright notice and either this complete permission notice or at +a minimum a reference to the UPL must be included in all copies or +substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. \ No newline at end of file diff --git a/src/oci-limits-mcp-server/README.md b/src/oci-limits-mcp-server/README.md new file mode 100644 index 0000000..8e29924 --- /dev/null +++ b/src/oci-limits-mcp-server/README.md @@ -0,0 +1,46 @@ +# OCI Limits MCP Server + +## Overview + +This MCP server exposes Oracle Cloud Infrastructure Limits APIs through tools mirroring public endpoints. + +## Running the Server + +To run the server: +```sh +uv run oracle.oci-limits-mcp-server +``` + +## Tools + +| Tool Name | Description | Parameters | +| --- | --- | --- | +| list_services | Returns supported services | compartment_id, sort_by='name', sort_order='ASC', limit=100, page=None, subscription_id=None | +| list_limit_definitions | Returns limit definitions | compartment_id, service_name=None, name=None, sort_by='name', sort_order='ASC', limit=100, page=None, subscription_id=None | +| list_limit_values | Returns limit values | compartment_id, service_name, scope_type=None, availability_domain=None, name=None, sort_by='name', sort_order='ASC', limit=100, page=None, subscription_id=None, external_location=None | +| get_resource_availability | Returns usage/availability for a specific limit | service_name, limit_name, compartment_id, availability_domain=None, subscription_id=None, external_location=None | + +## Authentication + +Follows the same pattern as other OCI MCP servers: +- Reads OCI config from ~/.oci/config +- Uses security token authentication +- Adds custom user agent header + +## Notes + +- For AD-scoped limits, `availability_domain` is required. +- Tools return dicts aligned with Swagger models. + +## License + +Copyright (c) 2025 Oracle and/or its affiliates. +Released under the Universal Permissive License v1.0. + +## Third-Party APIs + +Developers distributing binary implementations must obtain and provide required third-party licenses and copyright notices. + +## Disclaimer + +Users are responsible for local environment and credential safety. Different language models may yield different results. diff --git a/src/oci-limits-mcp-server/oracle/__init__.py b/src/oci-limits-mcp-server/oracle/__init__.py new file mode 100644 index 0000000..d9dff09 --- /dev/null +++ b/src/oci-limits-mcp-server/oracle/__init__.py @@ -0,0 +1,5 @@ +""" +Copyright (c) 2025, Oracle and/or its affiliates. +Licensed under the Universal Permissive License v1.0 as shown at +https://oss.oracle.com/licenses/upl. +""" diff --git a/src/oci-limits-mcp-server/oracle/oci_limits_mcp_server/__init__.py b/src/oci-limits-mcp-server/oracle/oci_limits_mcp_server/__init__.py new file mode 100644 index 0000000..fc04a8c --- /dev/null +++ b/src/oci-limits-mcp-server/oracle/oci_limits_mcp_server/__init__.py @@ -0,0 +1,8 @@ +""" +Copyright (c) 2025, Oracle and/or its affiliates. +Licensed under the Universal Permissive License v1.0 as shown at +https://oss.oracle.com/licenses/upl. +""" + +__project__ = "oracle.oci-limits-mcp-server" +__version__ = "1.0.1" diff --git a/src/oci-limits-mcp-server/oracle/oci_limits_mcp_server/server.py b/src/oci-limits-mcp-server/oracle/oci_limits_mcp_server/server.py new file mode 100644 index 0000000..9c61d34 --- /dev/null +++ b/src/oci-limits-mcp-server/oracle/oci_limits_mcp_server/server.py @@ -0,0 +1,331 @@ +""" +Copyright (c) 2025, Oracle and/or its affiliates. +Licensed under the Universal Permissive License v1.0 as shown at +https://oss.oracle.com/licenses/upl. +""" + +import os +from logging import Logger +from typing import Annotated, Optional + +import oci +from fastmcp import FastMCP + +from . import __project__, __version__ +from .utils import ( + list_limit_definitions_with_pagination, + list_limit_values_with_pagination, + list_services_with_pagination, +) + +logger = Logger(__name__, level="INFO") + +mcp = FastMCP(name=__project__) + + +def get_limits_client(): + """ + Build an OCI LimitsClient using Security Token auth (consistent with other servers in this repo). + Honors OCI_CONFIG_PROFILE if set. Adds a product-specific user agent. + """ + config = oci.config.from_file( + profile_name=os.getenv("OCI_CONFIG_PROFILE", oci.config.DEFAULT_PROFILE) + ) + user_agent_name = __project__.split("oracle.", 1)[1].split("-server", 1)[0] + config["additional_user_agent"] = f"{user_agent_name}/{__version__}" + + # Security token signer (same pattern as compute/identity) + private_key = oci.signer.load_private_key_from_file(config["key_file"]) + token_file = config["security_token_file"] + with open(token_file, "r") as f: + token = f.read() + signer = oci.auth.signers.SecurityTokenSigner(token, private_key) + + # Limits client + return oci.limits.LimitsClient(config, signer=signer) + + +def get_identity_client(): + """ + Build an OCI IdentityClient using Security Token auth (consistent with other servers in this repo). + Honors OCI_CONFIG_PROFILE if set. Adds a product-specific user agent. + """ + config = oci.config.from_file( + profile_name=os.getenv("OCI_CONFIG_PROFILE", oci.config.DEFAULT_PROFILE) + ) + user_agent_name = __project__.split("oracle.", 1)[1].split("-server", 1)[0] + config["additional_user_agent"] = f"{user_agent_name}/{__version__}" + + # Security token signer (same pattern as compute/identity) + private_key = oci.signer.load_private_key_from_file(config["key_file"]) + token_file = config["security_token_file"] + with open(token_file, "r") as f: + token = f.read() + signer = oci.auth.signers.SecurityTokenSigner(token, private_key) + + # Identity client + return oci.identity.IdentityClient(config, signer=signer) + + +# ---------------------------- +# Mappers to stable dict shapes +# ---------------------------- +def map_service_summary(svc: "oci.limits.models.ServiceSummary") -> dict: + return { + "name": getattr(svc, "name", None), + "description": getattr(svc, "description", None), + "supported_subscriptions": getattr(svc, "supported_subscriptions", None), + } + + +def map_limit_definition_summary( + defn: "oci.limits.models.LimitDefinitionSummary", +) -> dict: + return { + "name": getattr(defn, "name", None), + "serviceName": getattr(defn, "service_name", None), + "description": getattr(defn, "description", None), + "scopeType": getattr(defn, "scope_type", None), + "areQuotasSupported": getattr(defn, "are_quotas_supported", None), + "isResourceAvailabilitySupported": getattr( + defn, "is_resource_availability_supported", None + ), + "isDeprecated": getattr(defn, "is_deprecated", None), + "isEligibleForLimitIncrease": getattr( + defn, "is_eligible_for_limit_increase", None + ), + "isDynamic": getattr(defn, "is_dynamic", None), + "externalLocationSupportedSubscriptions": getattr( + defn, "external_location_supported_subscriptions", None + ), + "supportedSubscriptions": getattr(defn, "supported_subscriptions", None), + "supportedQuotaFamilies": getattr(defn, "supported_quota_families", None), + } + + +def map_limit_value_summary(val: "oci.limits.models.LimitValueSummary") -> dict: + return { + "name": getattr(val, "name", None), + "scopeType": getattr(val, "scope_type", None), + "availabilityDomain": getattr(val, "availability_domain", None), + "value": getattr(val, "value", None), + } + + +def map_resource_availability(ra: "oci.limits.models.ResourceAvailability") -> dict: + return { + "used": getattr(ra, "used", None), + "available": getattr(ra, "available", None), + "fractionalUsage": getattr(ra, "fractional_usage", None), + "fractionalAvailability": getattr(ra, "fractional_availability", None), + "effectiveQuotaValue": getattr(ra, "effective_quota_value", None), + } + + +def list_availability_domains(compartment_id: str) -> list[dict]: + client = get_identity_client() + response = client.list_availability_domains(compartment_id=compartment_id) + data = response.data + return [{"name": ad.name, "id": ad.id} for ad in data] + + +# ---------------------------- +# Tools +# ---------------------------- + + +@mcp.tool( + description="List availability domains for a given compartment needed for limits" +) +def provide_availability_domains_for_limits( + compartment_id: Annotated[str, "OCID of the compartment"], +) -> list[dict]: + return list_availability_domains(compartment_id) + + +@mcp.tool( + description="Returns the list of supported services that have resource limits exposed" +) +def list_services( + compartment_id: Annotated[str, "OCID of the root compartment (tenancy)"], + sort_by: Annotated[str, "Sort field: name or description"] = "name", + sort_order: Annotated[str, "Sort order: ASC or DESC"] = "ASC", + limit: Annotated[Optional[int], "Max items per page (1-1000)"] = 100, + page: Annotated[Optional[str], "Pagination token from a previous call"] = None, + subscription_id: Annotated[Optional[str], "Subscription OCID filter"] = None, +) -> list[dict]: + """ + Maps to GET /20190729/services + """ + try: + client = get_limits_client() + services = list_services_with_pagination( + client, + compartment_id=compartment_id, + sort_by=sort_by, + sort_order=sort_order, + limit=limit, + page=page, + subscription_id=subscription_id, + ) + service_summary = [map_service_summary(svc) for svc in services] + return service_summary + except Exception as e: + logger.error(f"Error in list_services: {e}") + raise + + +@mcp.tool(description="Get the list of resource limit definitions for a service") +def list_limit_definitions( + compartment_id: Annotated[str, "OCID of the root compartment (tenancy)"], + service_name: Annotated[Optional[str], "Target service name filter"] = None, + name: Annotated[Optional[str], "Specific resource limit name filter"] = None, + sort_by: Annotated[str, "Sort field: name or description"] = "name", + sort_order: Annotated[str, "Sort order: ASC or DESC"] = "ASC", + limit: Annotated[Optional[int], "Max items per page (1-1000)"] = 100, + page: Annotated[Optional[str], "Pagination token from a previous call"] = None, + subscription_id: Annotated[Optional[str], "Subscription OCID filter"] = None, +) -> list[dict]: + """ + Maps to GET /20190729/limitDefinitions + """ + try: + client = get_limits_client() + items = list_limit_definitions_with_pagination( + client, + compartment_id=compartment_id, + service_name=service_name, + name=name, + sort_by=sort_by, + sort_order=sort_order, + limit=limit, + page=page, + subscription_id=subscription_id, + ) + return [map_limit_definition_summary(d) for d in items] + except Exception as e: + logger.error(f"Error in list_limit_definitions: {e}") + raise + + +@mcp.tool( + description="Get the full list of resource limit values for the given service" +) +def get_limit_value( + compartment_id: Annotated[str, "OCID of the root compartment (tenancy)"], + service_name: Annotated[str, "Target service name"], + name: Annotated[str, "Specific resource limit name filter"], + scope_type: Annotated[str, "Filter by scope type: GLOBAL, REGION, or AD"], + availability_domain: Annotated[ + Optional[str], "If scope_type is AD, filter by availability domain" + ] = None, + sort_by: Annotated[str, "Sort field: name"] = "name", + sort_order: Annotated[str, "Sort order: ASC or DESC"] = "ASC", + limit: Annotated[Optional[int], "Max items per page (1-1000)"] = 100, + page: Annotated[Optional[str], "Pagination token from a previous call"] = None, + subscription_id: Annotated[Optional[str], "Subscription OCID filter"] = None, +) -> list[dict]: + """ + Maps to GET /20190729/limitValues + """ + try: + client = get_limits_client() + items = list_limit_values_with_pagination( + client, + compartment_id=compartment_id, + service_name=service_name, + scope_type=scope_type, + availability_domain=availability_domain, + name=name, + sort_by=sort_by, + sort_order=sort_order, + limit=limit, + page=page, + subscription_id=subscription_id, + ) + return [map_limit_value_summary(d) for d in items] + except Exception as e: + logger.error(f"Error in get_limit_value: {e}") + raise + + +@mcp.tool( + description="Get the availability and usage within a compartment for a given resource limit" +) +def get_resource_availability( + service_name: Annotated[str, "Service name of the target limit"], + limit_name: Annotated[str, "Limit name"], + compartment_id: Annotated[str, "OCID of the compartment to evaluate"], + availability_domain: Annotated[ + Optional[str], + "Required if the limit scopeType is AD; omit otherwise. Example: 'US-ASHBURN-AD-1'", + ] = None, + subscription_id: Annotated[Optional[str], "Subscription OCID filter"] = None, +) -> list[dict]: + """ + Maps to GET /20190729/services/{serviceName}/limits/{limitName}/resourceAvailability + """ + try: + client = get_limits_client() + limits = list_limit_definitions_with_pagination( + client, + compartment_id=compartment_id, + service_name=service_name, + name=limit_name, + ) + if len(limits) == 0: + return [ + { + "message": f"Limit '{limit_name}' not found for service '{service_name}'" + } + ] + + limit_definition = limits[0] + if not limit_definition.is_resource_availability_supported: + return [ + { + "message": f"Resource availability not supported for limit '{limit_name}'. " + f"Consider calling get_limit_value to get the limit value." + } + ] + + if limit_definition.scope_type == "AD": + availability_domains = list_availability_domains(compartment_id) + resource_availability = [] + for ad in availability_domains: + response: oci.response.Response = client.get_resource_availability( + service_name=service_name, + limit_name=limit_name, + compartment_id=compartment_id, + availability_domain=ad["name"], + subscription_id=subscription_id, + ) + data: oci.limits.models.ResourceAvailability = response.data + resource_availability.append( + { + "availabilityDomain": ad["name"], + "resourceAvailability": map_resource_availability(data), + } + ) + return resource_availability + else: + response: oci.response.Response = client.get_resource_availability( + service_name=service_name, + limit_name=limit_name, + compartment_id=compartment_id, + availability_domain=availability_domain, + subscription_id=subscription_id, + ) + data: oci.limits.models.ResourceAvailability = response.data + return [map_resource_availability(data)] + except Exception as e: + logger.error(f"Error in get_resource_availability: {e}") + raise + + +def main() -> None: + mcp.run() + + +if __name__ == "__main__": + main() diff --git a/src/oci-limits-mcp-server/oracle/oci_limits_mcp_server/tests/test_limit_tool.py b/src/oci-limits-mcp-server/oracle/oci_limits_mcp_server/tests/test_limit_tool.py new file mode 100644 index 0000000..8860722 --- /dev/null +++ b/src/oci-limits-mcp-server/oracle/oci_limits_mcp_server/tests/test_limit_tool.py @@ -0,0 +1,226 @@ +""" +Copyright (c) 2025, Oracle and/or its affiliates. +Licensed under the Universal Permissive License v1.0 as shown at +https://oss.oracle.com/licenses/upl. +""" + +from unittest.mock import MagicMock, create_autospec, patch + +import oci +import pytest +from fastmcp import Client +from oracle.oci_limits_mcp_server.server import mcp + + +class TestLimitsTools: + @pytest.mark.asyncio + @patch("oracle.oci_limits_mcp_server.server.get_identity_client") + async def test_provide_availability_domains_for_limits(self, mock_get_client): + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + mock_response = create_autospec(oci.response.Response) + mock_response.data = [ + oci.identity.models.AvailabilityDomain(name="AD1", id="US-ASHBURN-AD-1") + ] + mock_response.has_next_page = False + mock_response.next_page = None + mock_client.list_availability_domains.return_value = mock_response + + async with Client(mcp) as client: + result = ( + await client.call_tool( + "provide_availability_domains_for_limits", + { + "compartment_id": "ocid1.compartment.oc1..xxxx", + }, + ) + ).structured_content["result"] + + assert len(result) == 1 + assert result[0]["name"] == "AD1" + + @pytest.mark.asyncio + @patch("oracle.oci_limits_mcp_server.server.get_limits_client") + async def test_list_services(self, mock_get_client): + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + mock_response = create_autospec(oci.response.Response) + mock_response.data = [ + oci.limits.models.ServiceSummary(name="service1", description="Service 1") + ] + mock_response.has_next_page = False + mock_response.next_page = None + mock_client.list_services.return_value = mock_response + + async with Client(mcp) as client: + result = ( + await client.call_tool( + "list_services", + { + "compartment_id": "ocid1.compartment.oc1..xxxx", + }, + ) + ).structured_content["result"] + + assert len(result) == 1 + assert result[0]["name"] == "service1" + + @pytest.mark.asyncio + @patch("oracle.oci_limits_mcp_server.server.get_limits_client") + async def test_list_limit_definitions(self, mock_get_client): + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + mock_response = create_autospec(oci.response.Response) + mock_response.data = [ + oci.limits.models.LimitDefinitionSummary( + name="limit1", service_name="service1", description="Limit 1" + ) + ] + mock_response.has_next_page = False + mock_response.next_page = None + mock_client.list_limit_definitions.return_value = mock_response + + async with Client(mcp) as client: + result = ( + await client.call_tool( + "list_limit_definitions", + { + "compartment_id": "ocid1.compartment.oc1..xxxx", + }, + ) + ).structured_content["result"] + + assert len(result) == 1 + assert result[0]["name"] == "limit1" + + @pytest.mark.asyncio + @patch("oracle.oci_limits_mcp_server.server.get_limits_client") + async def test_get_limit_value(self, mock_get_client): + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + mock_response = create_autospec(oci.response.Response) + mock_response.data = [ + oci.limits.models.LimitValueSummary( + name="limit_value1", scope_type="GLOBAL", value=10 + ) + ] + mock_response.has_next_page = False + mock_response.next_page = None + mock_client.list_limit_values.return_value = mock_response + + async with Client(mcp) as client: + result = ( + await client.call_tool( + "get_limit_value", + { + "compartment_id": "ocid1.compartment.oc1..xxxx", + "service_name": "service1", + "name": "limit_value1", + "scope_type": "GLOBAL", + }, + ) + ).structured_content["result"] + + assert len(result) == 1 + assert result[0]["name"] == "limit_value1" + + @pytest.mark.asyncio + @patch("oracle.oci_limits_mcp_server.server.get_limits_client") + @patch("oracle.oci_limits_mcp_server.server.list_availability_domains") + async def test_get_resource_availability_ad_scope( + self, mock_list_ad, mock_get_client + ): + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + mock_limit_definition = create_autospec( + oci.limits.models.LimitDefinitionSummary + ) + mock_limit_definition.is_resource_availability_supported = True + mock_limit_definition.scope_type = "AD" + + mock_response = create_autospec(oci.response.Response) + mock_response.data = [mock_limit_definition] + mock_response.has_next_page = False + mock_response.next_page = None + mock_client.list_limit_definitions.return_value = mock_response + + mock_list_ad.return_value = [{"name": "AD1"}, {"name": "AD2"}] + + mock_response_ad1 = create_autospec(oci.response.Response) + mock_response_ad1.data = oci.limits.models.ResourceAvailability( + used=10, available=100 + ) + mock_response_ad2 = create_autospec(oci.response.Response) + mock_response_ad2.data = oci.limits.models.ResourceAvailability( + used=20, available=200 + ) + + mock_client.get_resource_availability.side_effect = [ + mock_response_ad1, + mock_response_ad2, + ] + + async with Client(mcp) as client: + result = await client.call_tool( + "get_resource_availability", + { + "service_name": "service1", + "limit_name": "limit1", + "compartment_id": "ocid1.compartment.oc1..xxxx", + }, + ) + + assert len(result.structured_content["result"]) == 2 + assert result.structured_content["result"][0]["availabilityDomain"] == "AD1" + assert ( + result.structured_content["result"][0]["resourceAvailability"]["used"] + == 10 + ) + assert result.structured_content["result"][1]["availabilityDomain"] == "AD2" + assert ( + result.structured_content["result"][1]["resourceAvailability"]["used"] + == 20 + ) + + @pytest.mark.asyncio + @patch("oracle.oci_limits_mcp_server.server.get_limits_client") + async def test_get_resource_availability_non_ad_scope(self, mock_get_client): + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + mock_limit_definition = create_autospec( + oci.limits.models.LimitDefinitionSummary + ) + mock_limit_definition.is_resource_availability_supported = True + mock_limit_definition.scope_type = "REGION" + + mock_response = create_autospec(oci.response.Response) + mock_response.data = [mock_limit_definition] + mock_response.has_next_page = False + mock_response.next_page = None + mock_client.list_limit_definitions.return_value = mock_response + + mock_response = create_autospec(oci.response.Response) + mock_response.data = oci.limits.models.ResourceAvailability( + used=10, available=100 + ) + mock_client.get_resource_availability.return_value = mock_response + + async with Client(mcp) as client: + result = await client.call_tool( + "get_resource_availability", + { + "service_name": "service1", + "limit_name": "limit1", + "compartment_id": "ocid1.compartment.oc1..xxxx", + }, + ) + + assert len(result.structured_content["result"]) == 1 + assert result.structured_content["result"][0]["used"] == 10 + assert result.structured_content["result"][0]["available"] == 100 diff --git a/src/oci-limits-mcp-server/oracle/oci_limits_mcp_server/utils.py b/src/oci-limits-mcp-server/oracle/oci_limits_mcp_server/utils.py new file mode 100644 index 0000000..82a232e --- /dev/null +++ b/src/oci-limits-mcp-server/oracle/oci_limits_mcp_server/utils.py @@ -0,0 +1,129 @@ +from logging import Logger +from typing import List, Optional + +import oci + +logger = Logger(__name__, level="INFO") + + +def list_services_with_pagination( + client: oci.limits.LimitsClient, + compartment_id: str, + sort_by: str = "name", + sort_order: str = "ASC", + limit: Optional[int] = 100, + page: Optional[str] = None, + subscription_id: Optional[str] = None, +) -> List[oci.limits.models.ServiceSummary]: + try: + items: List[oci.limits.models.ServiceSummary] = [] + next_page = page + has_next_page = True + + while has_next_page: + response = client.list_services( + compartment_id=compartment_id, + sort_by=sort_by, + sort_order=sort_order, + limit=limit, + page=next_page, + subscription_id=subscription_id, + ) + data = response.data or [] + items.extend(data) + has_next_page = response.has_next_page + next_page = response.next_page + + if page is not None: + break + + return items + except Exception as e: + logger.error(f"Error in list_services: {e}") + raise + + +def list_limit_definitions_with_pagination( + client: oci.limits.LimitsClient, + compartment_id: str, + service_name: Optional[str] = None, + name: Optional[str] = None, + sort_by: str = "name", + sort_order: str = "ASC", + limit: Optional[int] = 100, + page: Optional[str] = None, + subscription_id: Optional[str] = None, +) -> List[oci.limits.models.LimitDefinitionSummary]: + try: + items: List[oci.limits.models.LimitDefinitionSummary] = [] + next_page = page + has_next_page = True + + while has_next_page: + response = client.list_limit_definitions( + compartment_id=compartment_id, + service_name=service_name, + name=name, + sort_by=sort_by, + sort_order=sort_order, + limit=limit, + page=next_page, + subscription_id=subscription_id, + ) + data = response.data or [] + items.extend(data) + has_next_page = response.has_next_page + next_page = response.next_page + + if page is not None: + break + + return items + except Exception as e: + logger.error(f"Error in list_limit_definitions: {e}") + raise + + +def list_limit_values_with_pagination( + client: oci.limits.LimitsClient, + compartment_id: str, + service_name: str, + scope_type: str, + availability_domain: Optional[str] = None, + name: Optional[str] = None, + sort_by: str = "name", + sort_order: str = "ASC", + limit: Optional[int] = 100, + page: Optional[str] = None, + subscription_id: Optional[str] = None, +) -> List[oci.limits.models.LimitValueSummary]: + try: + items: List[oci.limits.models.LimitValueSummary] = [] + next_page = page + has_next_page = True + + while has_next_page: + response = client.list_limit_values( + compartment_id=compartment_id, + service_name=service_name, + scope_type=scope_type, + availability_domain=availability_domain, + name=name, + sort_by=sort_by, + sort_order=sort_order, + limit=limit, + page=next_page, + subscription_id=subscription_id, + ) + data = response.data or [] + items.extend(data) + has_next_page = response.has_next_page + next_page = response.next_page + + if page is not None: + break + + return items + except Exception as e: + logger.error(f"Error in list_limit_values: {e}") + raise diff --git a/src/oci-limits-mcp-server/pyproject.toml b/src/oci-limits-mcp-server/pyproject.toml new file mode 100644 index 0000000..113ae7d --- /dev/null +++ b/src/oci-limits-mcp-server/pyproject.toml @@ -0,0 +1,36 @@ +[project] +name = "oracle.oci-limits-mcp-server" +version = "1.0.1" +description = "OCI Limits MCP server" +readme = "README.md" +requires-python = ">=3.13" +authors = [ + {name = "Oracle MCP", email = "237432095+oracle-mcp@users.noreply.github.com"}, +] +dependencies = [ + "fastmcp==2.12.2", + "oci==2.160.0" +] + +classifiers = [ + "License :: OSI Approved :: Universal Permissive License (UPL)", + "Operating System :: OS Independent", + "Programming Language :: Python", + "Programming Language :: Python :: 3.13", +] + +[project.scripts] +"oracle.oci-limits-mcp-server" = "oracle.oci_limits_mcp_server.server:main" + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[tool.hatch.build.targets.wheel] +packages = ["oracle"] + +[dependency-groups] +dev = [ + "pytest>=8.4.2", + "pytest-asyncio>=1.2.0", +]