FAAASCON-3029 - Add mcp tool to list service attachments

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"),
+    )

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()
 

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"