Merge pull request #21 from shibbirmcc/#15-implement-kong-plugins-management-tools

#15 Implement Kong Plugins Management Tools #15

Author: Mohamad-Bahmani

README.md

@@ -164,7 +164,11 @@ docker run -p 9000:9000 -e FASTMCP_PORT=9000 kong-mcp-server
   - `kong_create_route`: Create new route
   - `kong_update_route`: Update existing route
   - `kong_delete_route`: Delete route
-
+- **Kong Plugins**: Management and retrieval of Kong plugins with filtering and scoping support  
+  - `kong_get_plugins`: Retrieve all plugins with optional filtering and pagination  
+  - `kong_get_plugins_by_service`: Retrieve plugins scoped to a specific service  
+  - `kong_get_plugins_by_route`: Retrieve plugins scoped to a specific route  
+  - `kong_get_plugins_by_consumer`: Retrieve plugins scoped to a specific consumer  
 ### Adding New Tools
 
 1. Create a new module in `src/kong_mcp_server/tools/`

src/kong_mcp_server/kong_client.py

@@ -338,7 +338,7 @@ async def delete_route(self, route_id: str) -> None:
         """
         await self.delete(f"/routes/{route_id}")
 
-    async def get_plugins(self, **params: Any) -> List[Dict[str, Any]]:
+    async def get_plugins_as_list(self, **params: Any) -> List[Dict[str, Any]]:
         """Get all Kong plugins.
 
         Args:
@@ -401,3 +401,56 @@ async def health_check(self) -> Dict[str, Any]:
             Kong status information
         """
         return await self.get("/status")
+
+    async def get_plugins(
+        self, params: Optional[Dict[str, Any]] = None
+    ) -> Dict[str, Any]:
+        """
+        GET /plugins with optional query params.
+        Returns the full Kong pagination envelope: {"data": [...], "offset": "..."}
+        """
+        return await self._request(
+            "GET",
+            "/plugins",
+            params=params,
+            json_data=None,
+        )
+
+    async def get_plugins_by_service(
+        self, service_id: str, params: Optional[Dict[str, Any]] = None
+    ) -> Dict[str, Any]:
+        """
+        GET /plugins by service id.
+        Returns the full Kong pagination envelope: {"data": [...], "offset": "..."}
+        """
+        return await self._request(
+            "GET",
+            f"/services/{service_id}/plugins",
+            params=params,
+            json_data=None,
+        )
+
+    async def get_plugins_by_route(
+        self, route_id: str, params: Optional[Dict[str, Any]] = None
+    ) -> Dict[str, Any]:
+        """
+        GET /plugins by route id.
+        Returns the full Kong pagination envelope: {"data": [...], "offset": "..."}
+        """
+        return await self._request(
+            "GET",
+            f"/routes/{route_id}/plugins",
+            params=params,
+            json_data=None,
+        )
+
+    async def get_plugins_by_consumer(
+        self, consumer_id: str, params: Optional[Dict[str, Any]] = None
+    ) -> Dict[str, Any]:
+        """
+        GET /plugins by consumer id.
+        Returns the full Kong pagination envelope: {"data": [...], "offset": "..."}
+        """
+        return await self._request(
+            "GET", f"/consumers/{consumer_id}/plugins", params=params, json_data=None
+        )

src/kong_mcp_server/tools/kong_plugins.py

@@ -0,0 +1,122 @@
+"""Kong plugins management tools."""
+
+from typing import Any, Dict, Optional
+
+from kong_mcp_server.kong_client import KongClient
+
+
+async def get_plugins(
+    name: Optional[str] = None, offset: Optional[str] = None, size: Optional[int] = None
+) -> Dict[str, Any]:
+    """Retrieve Kong plugins.
+
+    Args:
+    name: Filter by plugin name (e.g., "rate-limiting", "rate-limiting-advanced")
+    offset: Pagination cursor returned by Kong from a previous call
+    size: Page size (Kong default is typically 100)
+
+    Returns:
+        List of Kong plugins data.
+    """
+    if size is not None:
+        if size < 1 or size > 1000:
+            raise ValueError("Size must be between 1 and 1000")
+
+    params: Dict[str, Any] = {}
+    if name:
+        params["name"] = name
+    if offset:
+        params["offset"] = offset
+    if size:
+        params["size"] = size
+
+    async with KongClient() as client:
+        response = await client.get_plugins(params=params)
+    next_offset = response.get("offset")
+    if not next_offset and "next" in response:
+        next_offset = response.get("next")
+
+    return {
+        "data": response.get("data", []),
+        "offset": next_offset,
+    }
+
+
+async def get_plugins_by_service(
+    service_id: str, size: Optional[int] = None, offset: Optional[str] = None
+) -> dict[str, Any]:
+    """
+    Retrieve plugins associated with a specific Kong service.
+
+    Args:
+        service_id: The ID (or name) of the Kong service.
+        size: Optional number of plugins to return.
+        offset: Optional pagination cursor.
+
+    Returns:
+        A dictionary with:
+        - data: List of plugin objects
+        - next: URL for next page if pagination is present
+    """
+    params: Dict[str, Any] = {}
+    if size:
+        params["size"] = size
+    if offset:
+        params["offset"] = offset
+
+    async with KongClient() as client:
+        return await client.get_plugins_by_service(service_id=service_id, params=params)
+
+
+async def get_plugins_by_route(
+    route_id: str, size: Optional[int] = None, offset: Optional[str] = None
+) -> dict[str, Any]:
+    """
+    Retrieve plugins associated with a specific Kong route.
+
+    Args:
+        route_id: The ID (or name) of the Kong route.
+        size: Optional number of plugins to return.
+        offset: Optional pagination cursor.
+
+    Returns:
+        A dictionary with:
+        - data: List of plugin objects
+        - next: URL for next page if pagination is present
+    """
+    params: Dict[str, Any] = {}
+    if size:
+        params["size"] = size
+    if offset:
+        params["offset"] = offset
+
+    async with KongClient() as client:
+        return await client.get_plugins_by_route(route_id=route_id, params=params)
+
+
+async def get_plugins_by_consumer(
+    consumer_id: str, size: Optional[int] = None, offset: Optional[str] = None
+) -> dict[str, Any]:
+    """
+    Retrieve plugins associated with a specific Kong consumer.
+
+    Args:
+        consumer_id: The ID (or name) of the Kong consumer.
+        size: Optional number of plugins to return.
+        offset: Optional pagination cursor.
+
+    Returns:
+        A dictionary with:
+        - data: List of plugin objects
+        - next: URL for next page if pagination is present
+    """
+    params: Dict[str, Any] = {}
+    if size:
+        params["size"] = size
+    if offset:
+        params["offset"] = offset
+
+    async with KongClient() as client:
+        return await client.get_plugins_by_consumer(
+            consumer_id=consumer_id, params=params
+        )

src/kong_mcp_server/tools_config.json

@@ -55,6 +55,34 @@
       "module": "kong_mcp_server.tools.kong_routes",
       "function": "delete_route",
       "enabled": true
+    },
+    "kong_get_plugins": {
+      "name": "kong_get_plugins",
+      "description": "Retrieve global Kong plugins with optional filtering by name and pagination.",
+      "module": "kong_mcp_server.tools.kong_plugins",
+      "function": "get_plugins",
+      "enabled": true
+    },
+    "kong_get_plugins_by_service": {
+      "name": "kong_get_plugins_by_service",
+      "description": "Retrieve service scoped Kong plugins with optional filtering by name and pagination.",
+      "module": "kong_mcp_server.tools.kong_plugins",
+      "function": "get_plugins_by_service",
+      "enabled": true
+    },
+      "kong_get_plugins_by_route": {
+      "name": "kong_get_plugins_by_route",
+      "description": "Retrieve route scoped Kong plugins with optional filtering by name and pagination.",
+      "module": "kong_mcp_server.tools.kong_plugins",
+      "function": "get_plugins_by_route",
+      "enabled": true
+    },
+      "kong_get_plugins_by_consumer": {
+      "name": "kong_get_plugins_by_consumer",
+      "description": "Retrieve consumer scoped Kong plugins with optional filtering by name and pagination.",
+      "module": "kong_mcp_server.tools.kong_plugins",
+      "function": "get_plugins_by_consumer",
+      "enabled": true
     }
   }
 }

tests/test_kong_client.py

@@ -481,7 +481,7 @@ async def test_get_plugins(self, config: KongClientConfig) -> None:
         mock_response = {"data": [{"id": "1", "name": "rate-limiting"}]}
 
         with patch.object(client, "get", return_value=mock_response) as mock_get:
-            result = await client.get_plugins()
+            result = await client.get_plugins_as_list()
 
             assert result == [{"id": "1", "name": "rate-limiting"}]
             mock_get.assert_called_once_with("/plugins", params={})

tests/test_kong_integration.py

@@ -57,14 +57,51 @@ def mock_kong_responses(self) -> Dict[str, Any]:
             "plugins": {
                 "data": [
                     {
-                        "id": "plugin-1",
+                        "id": "plugin-global-1",
+                        "name": "cors",
+                        "config": {},
+                        "created_at": 1618846400,
+                    }
+                ],
+                "offset": None,
+            },
+            "plugins_by_service": {
+                "data": [
+                    {
+                        "id": "plugin-service-1",
                         "name": "rate-limiting",
-                        "config": {"minute": 100, "hour": 1000},
                         "service": {"id": "service-1"},
+                        "config": {"minute": 100},
                         "created_at": 1618846400,
                     }
-                ]
+                ],
+                "offset": None,
+            },
+            "plugins_by_route": {
+                "data": [
+                    {
+                        "id": "plugin-route-1",
+                        "name": "jwt",
+                        "route": {"id": "route-1"},
+                        "config": {"key": "value"},
+                        "created_at": 1618846400,
+                    }
+                ],
+                "offset": None,
             },
+            "plugins_by_consumer": {
+                "data": [
+                    {
+                        "id": "plugin-consumer-1",
+                        "name": "key-auth",
+                        "consumer": {"id": "consumer-1"},
+                        "config": {},
+                        "created_at": 1618846400,
+                    }
+                ],
+                "offset": None,
+            },
+            "consumers": {"data": [{"id": "consumer-1", "username": "test-consumer"}]},
             "status": {
                 "database": {"reachable": True},
                 "server": {"connections_accepted": 100},
@@ -213,6 +250,93 @@ async def test_kong_client_route_operations_integration(
                 assert routes[0]["name"] == "test-route"
                 mock_request.assert_called_with("GET", "/routes", params={"size": 10})
 
+    @pytest.mark.asyncio
+    async def test_kong_client_plugins_operations_integration(
+        self, mock_kong_responses: Dict[str, Any]
+    ) -> None:
+        config = KongClientConfig(base_url="http://kong-admin:8001")
+        async with KongClient(config) as client:
+            with patch.object(client, "_request") as mock_request:
+                mock_response = mock_kong_responses["plugins"]
+                mock_request.return_value = mock_response
+
+                result = await client.get_plugins(params={"name": "jwt", "size": 10})
+
+                assert result["data"] == mock_response["data"]
+                assert result["offset"] == mock_response.get(
+                    "offset"
+                ) or mock_response.get("next")
+                # Validate call parameters
+                mock_request.assert_called_with(
+                    "GET",
+                    "/plugins",
+                    params={"name": "jwt", "size": 10},
+                    json_data=None,
+                )
+
+    async def test_get_plugins_by_service(self, mock_kong_responses: Dict[str, Any]):
+        """Test get_plugins_by_service tool."""
+        config = KongClientConfig(base_url="http://kong-admin:8001")
+        async with KongClient(config) as client:
+            with patch.object(client, "_request") as mock_request:
+                service_id = "service-1"
+                mock_response = mock_kong_responses["plugins_by_service"]
+                mock_request.return_value = mock_response
+
+                result = await client.get_plugins_by_service(
+                    service_id, params={"size": 5}
+                )
+
+                assert result["data"] == mock_response["data"]
+                mock_request.assert_called_with(
+                    "GET",
+                    f"/services/{service_id}/plugins",
+                    params={"size": 5},
+                    json_data=None,
+                )
+
+    async def test_get_plugins_by_route(self, mock_kong_responses: Dict[str, Any]):
+        """Test get_plugins_by_route tool."""
+        config = KongClientConfig(base_url="http://kong-admin:8001")
+        async with KongClient(config) as client:
+            with patch.object(client, "_request") as mock_request:
+                route_id = "route-1"
+                mock_response = mock_kong_responses["plugins_by_route"]
+                mock_request.return_value = mock_response
+
+                result = await client.get_plugins_by_route(
+                    route_id, params={"offset": "cursor0"}
+                )
+
+                assert result["data"] == mock_response["data"]
+                mock_request.assert_called_with(
+                    "GET",
+                    f"/routes/{route_id}/plugins",
+                    params={"offset": "cursor0"},
+                    json_data=None,
+                )
+
+    async def test_get_plugins_by_consumer(self, mock_kong_responses: Dict[str, Any]):
+        """Test get_plugins_by_consumer tool."""
+        config = KongClientConfig(base_url="http://kong-admin:8001")
+        async with KongClient(config) as client:
+            with patch.object(client, "_request") as mock_request:
+                consumer_id = "consumer-1"
+                mock_response = mock_kong_responses["plugins_by_consumer"]
+                mock_request.return_value = mock_response
+
+                result = await client.get_plugins_by_consumer(
+                    consumer_id, params={"size": 3}
+                )
+
+                assert result["data"] == mock_response["data"]
+                mock_request.assert_called_with(
+                    "GET",
+                    f"/consumers/{consumer_id}/plugins",
+                    params={"size": 3},
+                    json_data=None,
+                )
+
     @pytest.mark.asyncio
     async def test_kong_client_error_handling_integration(self) -> None:
         """Test Kong client error handling in integration scenarios."""