From 2cd4ed0e0532eb508f9ee3729b66f6f15f6d7442 Mon Sep 17 00:00:00 2001 From: Tirthankar Nayak Date: Wed, 14 Jan 2026 02:51:33 +0530 Subject: [PATCH] FAAASCON-3005 - Add MCP tool to list environment inside family, FAAASCON-3029 - Add mcp tool to list service attachments --- .../oracle/oci_faaas_mcp_server/models.py | 73 +++++++++++++++++ .../oracle/oci_faaas_mcp_server/server.py | 81 +++++++++++++++++++ .../tests/test_faaas_tools.py | 31 +++++++ 3 files changed, 185 insertions(+) diff --git a/src/oci-faaas-mcp-server/oracle/oci_faaas_mcp_server/models.py b/src/oci-faaas-mcp-server/oracle/oci_faaas_mcp_server/models.py index 10e95562..842c2ea9 100644 --- a/src/oci-faaas-mcp-server/oracle/oci_faaas_mcp_server/models.py +++ b/src/oci-faaas-mcp-server/oracle/oci_faaas_mcp_server/models.py @@ -7,6 +7,7 @@ from datetime import datetime from typing import Any, Dict, List, Optional +import oci from pydantic import BaseModel, Field @@ -137,6 +138,58 @@ class FusionEnvironmentStatus(BaseModel): ) +class ServiceAttachmentSummary(BaseModel): + """Pydantic model respresenting a Service Attachment.""" + + id: Optional[str] = Field( + None, description="Unique identifier that is immutable on creation." + ) + display_name: Optional[str] = Field( + None, description="ServiceInstance identifier; can be renamed." + ) + service_instance_type: Optional[str] = Field( + None, description="Type of the service." + ) + service_instance_id: Optional[str] = Field( + None, + description=( + "ID of the service instance that can be used to identify this on the service control plane." + ), + ) + service_url: Optional[str] = Field(None, description="Service URL of the instance.") + time_created: Optional[datetime] = Field( + None, description="The time the service instance was created (RFC3339)." + ) + time_updated: Optional[datetime] = Field( + None, description="The time the service instance was updated (RFC3339)." + ) + lifecycle_state: Optional[str] = Field( + None, description="The current state of the ServiceInstance." + ) + lifecycle_details: Optional[str] = Field( + None, + description=( + "Detailed message describing the current state. Useful for actionable information when Failed." + ), + ) + is_sku_based: Optional[bool] = Field( + None, + description=( + "Whether this service is provisioned due to subscription to a specific SKU." + ), + ) + freeform_tags: Optional[Dict[str, str]] = Field( + None, + description="Free-form tags for this resource as simple key/value pairs.", + ) + defined_tags: Optional[Dict[str, Dict[str, Any]]] = Field( + None, + description=( + "Defined tags for this resource. Each key is predefined and scoped to a namespace." + ), + ) + + def _get(data: Any, key: str) -> Any: """Safe getter to support both dicts and SDK objects.""" if isinstance(data, dict): @@ -213,3 +266,23 @@ def map_fusion_environment_status(data: Any) -> FusionEnvironmentStatus: time_created=_get(data, "time_created"), details=details or None, ) + + +def map_service_attachment_summary( + data: oci.fusion_apps.models.ServiceAttachmentSummary, +) -> ServiceAttachmentSummary: + """Map SDK model or dict to ServiceAttachmentSummary.""" + return ServiceAttachmentSummary( + id=_get(data, "id"), + display_name=_get(data, "display_name"), + service_instance_type=_get(data, "service_instance_type"), + service_instance_id=_get(data, "service_instance_id"), + service_url=_get(data, "service_url"), + time_created=_get(data, "time_created"), + time_updated=_get(data, "time_updated"), + lifecycle_state=_get(data, "lifecycle_state"), + lifecycle_details=_get(data, "lifecycle_details"), + is_sku_based=_get(data, "is_sku_based"), + freeform_tags=_get(data, "freeform_tags"), + defined_tags=_get(data, "defined_tags"), + ) diff --git a/src/oci-faaas-mcp-server/oracle/oci_faaas_mcp_server/server.py b/src/oci-faaas-mcp-server/oracle/oci_faaas_mcp_server/server.py index 2b88b778..8f41663f 100644 --- a/src/oci-faaas-mcp-server/oracle/oci_faaas_mcp_server/server.py +++ b/src/oci-faaas-mcp-server/oracle/oci_faaas_mcp_server/server.py @@ -17,9 +17,11 @@ FusionEnvironment, FusionEnvironmentFamily, FusionEnvironmentStatus, + ServiceAttachmentSummary, map_fusion_environment, map_fusion_environment_family, map_fusion_environment_status, + map_service_attachment_summary, ) logger = Logger(__name__, level="INFO") @@ -216,6 +218,85 @@ def get_fusion_environment_status( return map_fusion_environment_status(response.data) +@mcp.tool( + description=( + "Returns a list of Service attachments present under the specified Fusion environment " + ) +) +def list_service_attachments( + fusion_environment_id: str = Field(..., description="Fusion environment OCID"), + display_name: Optional[str] = Field( + None, description="Filter to match entire display name." + ), + lifecycle_state: Optional[ + Literal["CREATING", "UPDATING", "ACTIVE", "DELETING", "DELETED", "FAILED"] + ] = Field( + None, + description=( + "Filter by lifecycle state. Allowed: CREATING, UPDATING, ACTIVE, " + "DELETING, DELETED, FAILED" + ), + ), + service_instance_type: Optional[ + Literal[ + "DIGITAL_ASSISTANT", + "INTEGRATION_CLOUD", + "ANALYTICS_WAREHOUSE", + "VBCS", + "VISUAL_BUILDER_STUDIO", + ] + ] = Field( + None, + description=( + "Filter by service instance type. Allowed: " + "DIGITAL_ASSISTANT, INTEGRATION_CLOUD, ANALYTICS_WAREHOUSE, VBCS, VISUAL_BUILDER_STUDIO" + ), + ), +) -> list[ServiceAttachmentSummary]: + client = get_faaas_client() + + service_attachments: list[ServiceAttachmentSummary] = [] + next_page: Optional[str] = None + has_next_page = True + + while has_next_page: + kwargs: dict[str, Any] = {"fusion_environment_id": fusion_environment_id} + if next_page is not None: + kwargs["page"] = next_page + if display_name is not None: + kwargs["display_name"] = display_name + if lifecycle_state is not None: + kwargs["lifecycle_state"] = lifecycle_state + if service_instance_type is not None: + kwargs["service_instance_type"] = service_instance_type + + response: oci.response.Response = client.list_service_attachments(**kwargs) + + # Normalize response data to an iterable without using helpers + data_obj = response.data or [] + items = getattr(data_obj, "items", None) + iterable = ( + items + if items is not None + else (data_obj if isinstance(data_obj, list) else [data_obj]) + ) + for d in iterable: + service_attachments.append(map_service_attachment_summary(d)) + + # Robust pagination handling with header fallback + headers = getattr(response, "headers", None) + next_page = getattr(response, "next_page", None) + if next_page is None and headers: + try: + next_page = dict(headers).get("opc-next-page") + except Exception: + next_page = None + has_next_page = next_page is not None + + logger.info(f"Found {len(service_attachments)} Service attachements") + return service_attachments + + def main(): mcp.run() diff --git a/src/oci-faaas-mcp-server/oracle/oci_faaas_mcp_server/tests/test_faaas_tools.py b/src/oci-faaas-mcp-server/oracle/oci_faaas_mcp_server/tests/test_faaas_tools.py index 65bc7e23..9d77facc 100644 --- a/src/oci-faaas-mcp-server/oracle/oci_faaas_mcp_server/tests/test_faaas_tools.py +++ b/src/oci-faaas-mcp-server/oracle/oci_faaas_mcp_server/tests/test_faaas_tools.py @@ -146,6 +146,37 @@ async def test_get_fusion_environment_status(self, mock_get_client): assert result["fusion_environment_id"] == "env1" assert result["status"] == "ACTIVE" + @pytest.mark.asyncio + @patch("oracle.oci_faaas_mcp_server.server.get_faaas_client") + async def test_list_service_attachments(self, mock_get_client): + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + mock_list_response = create_autospec(oci.response.Response) + mock_list_response.data = [ + { + "id": "integration1", + "display_name": "Oracle Digital Assistant", + "lifecycle_state": "ACTIVE", + "service_instance_type": "DIGITAL_ASSISTANT", + } + ] + mock_list_response.has_next_page = False + mock_list_response.next_page = None + mock_client.list_service_attachments.return_value = mock_list_response + + async with Client(mcp) as client: + call_tool_result = await client.call_tool( + "list_service_attachments", + { + "fusion_environment_id": "ocid123", + }, + ) + result = call_tool_result.structured_content["result"] + + assert len(result) == 1 + assert result[0]["id"] == "integration1" + def test_get_faaas_client_initializes_client(self, tmp_path): # Prepare temp token file token_file = tmp_path / "token"