feat(network): add router CRUD tools (#60)

Author: platanus-kr

src/openstack_mcp_server/tools/network_tools.py

@@ -1,10 +1,15 @@
 from fastmcp import FastMCP
 
 from .base import get_openstack_conn
+from .request.network import (
+    ExternalGatewayInfo,
+    Route,
+)
 from .response.network import (
     FloatingIP,
     Network,
     Port,
+    Router,
     Subnet,
 )
 
@@ -80,6 +85,7 @@ def create_network(
         provider_network_type: str | None = None,
         provider_physical_network: str | None = None,
         provider_segmentation_id: int | None = None,
+        project_id: str | None = None,
     ) -> Network:
         """
         Create a new Network.
@@ -107,6 +113,9 @@ def create_network(
         if provider_network_type:
             network_args["provider_network_type"] = provider_network_type
 
+        if project_id:
+            network_args["project_id"] = project_id
+
         if provider_physical_network:
             network_args["provider_physical_network"] = (
                 provider_physical_network
@@ -860,3 +869,176 @@ def _convert_to_floating_ip_model(self, openstack_ip) -> FloatingIP:
             port_id=openstack_ip.port_id,
             router_id=openstack_ip.router_id,
         )
+
+    def get_routers(
+        self,
+        status_filter: str | None = None,
+        project_id: str | None = None,
+        is_admin_state_up: bool | None = None,
+    ) -> list[Router]:
+        """
+        Get the list of Routers with optional filtering.
+        :param status_filter: Filter by router status (e.g., `ACTIVE`, `DOWN`)
+        :param project_id: Filter by project ID
+        :param is_admin_state_up: Filter by admin state
+        :return: List of Router objects
+        """
+        conn = get_openstack_conn()
+        filters: dict = {}
+        if status_filter:
+            filters["status"] = status_filter.upper()
+        if project_id:
+            filters["project_id"] = project_id
+        if is_admin_state_up is not None:
+            filters["admin_state_up"] = is_admin_state_up
+        routers = conn.list_routers(filters=filters)
+        return [self._convert_to_router_model(r) for r in routers]
+
+    def create_router(
+        self,
+        name: str | None = None,
+        description: str | None = None,
+        is_admin_state_up: bool = True,
+        is_distributed: bool | None = None,
+        project_id: str | None = None,
+        external_gateway_info: ExternalGatewayInfo | None = None,
+    ) -> Router:
+        """
+        Create a new Router.
+        Typical use-cases:
+        - Create basic router: name="r1" (defaults to admin_state_up=True)
+        - Create distributed router: is_distributed=True
+        - Create with external gateway for north-south traffic:
+          external_gateway_info={"network_id": "ext-net", "enable_snat": True,
+          "external_fixed_ips": [{"subnet_id": "ext-subnet", "ip_address": "203.0.113.10"}]}
+        - Create with project ownership: project_id="proj-1"
+        Notes:
+        - external_gateway_info should follow Neutron schema: at minimum include
+          "network_id"; optional keys include "enable_snat" and "external_fixed_ips".
+        :param name: Router name
+        :param description: Router description
+        :param is_admin_state_up: Administrative state
+        :param is_distributed: Distributed router flag
+        :param project_id: Project ownership
+        :param external_gateway_info: External gateway info dict
+        :return: Created Router object
+        """
+        conn = get_openstack_conn()
+        router_args: dict = {"admin_state_up": is_admin_state_up}
+        if name is not None:
+            router_args["name"] = name
+        if description is not None:
+            router_args["description"] = description
+        if is_distributed is not None:
+            router_args["distributed"] = is_distributed
+        if project_id is not None:
+            router_args["project_id"] = project_id
+        if external_gateway_info is not None:
+            router_args["external_gateway_info"] = (
+                external_gateway_info.model_dump(exclude_none=True)
+            )
+        router = conn.network.create_router(**router_args)
+        return self._convert_to_router_model(router)
+
+    def get_router_detail(self, router_id: str) -> Router:
+        """
+        Get detailed information about a specific Router.
+        :param router_id: ID of the router to retrieve
+        :return: Router details
+        """
+        conn = get_openstack_conn()
+        router = conn.network.get_router(router_id)
+        return self._convert_to_router_model(router)
+
+    def update_router(
+        self,
+        router_id: str,
+        name: str | None = None,
+        description: str | None = None,
+        is_admin_state_up: bool | None = None,
+        is_distributed: bool | None = None,
+        external_gateway_info: ExternalGatewayInfo | None = None,
+        clear_external_gateway: bool = False,
+        routes: list[Route] | None = None,
+    ) -> Router:
+        """
+        Update Router attributes atomically. Only provided parameters are changed;
+        omitted parameters remain untouched.
+        Typical use-cases:
+        - Rename and change description: name="r-new", description="d".
+        - Toggle admin state: read current via get_router_detail(); pass inverted bool to is_admin_state_up.
+        - Set distributed flag: is_distributed=True or False.
+        - Set external gateway: external_gateway_info={"network_id": "ext-net", "enable_snat": True, "external_fixed_ips": [...]}.
+        - Clear external gateway: clear_external_gateway=True (takes precedence over external_gateway_info).
+        - Replace static routes: routes=[{"destination": "192.0.2.0/24", "nexthop": "10.0.0.1"}]. Pass [] to remove all routes.
+        Notes:
+        - For list-typed fields (routes), the provided list replaces the entire list on the server.
+        - To clear external gateway, use clear_external_gateway=True. If both provided, clear_external_gateway takes precedence.
+        :param router_id: ID of the router to update
+        :param name: New router name
+        :param description: New router description
+        :param is_admin_state_up: Administrative state
+        :param is_distributed: Distributed router flag
+        :param external_gateway_info: External gateway info dict to set
+        :param clear_external_gateway: If True, clear external gateway (set to None)
+        :param routes: Static routes (replaces entire list)
+        :return: Updated Router object
+        """
+        conn = get_openstack_conn()
+        update_args: dict = {}
+        if name is not None:
+            update_args["name"] = name
+        if description is not None:
+            update_args["description"] = description
+        if is_admin_state_up is not None:
+            update_args["admin_state_up"] = is_admin_state_up
+        if is_distributed is not None:
+            update_args["distributed"] = is_distributed
+        if clear_external_gateway:
+            update_args["external_gateway_info"] = None
+        elif external_gateway_info is not None:
+            update_args["external_gateway_info"] = (
+                external_gateway_info.model_dump(exclude_none=True)
+            )
+        if routes is not None:
+            update_args["routes"] = [
+                r.model_dump(exclude_none=True) for r in routes
+            ]
+        if not update_args:
+            current = conn.network.get_router(router_id)
+            return self._convert_to_router_model(current)
+        router = conn.network.update_router(router_id, **update_args)
+        return self._convert_to_router_model(router)
+
+    def delete_router(self, router_id: str) -> None:
+        """
+        Delete a Router.
+        :param router_id: ID of the router to delete
+        :return: None
+        """
+        conn = get_openstack_conn()
+        conn.network.delete_router(router_id, ignore_missing=False)
+        return None
+
+    def _convert_to_router_model(self, openstack_router) -> Router:
+        """
+        Convert an OpenStack Router object to a Router pydantic model.
+        :param openstack_router: OpenStack router object
+        :return: Pydantic Router model
+        """
+        return Router(
+            id=openstack_router.id,
+            name=getattr(openstack_router, "name", None),
+            status=getattr(openstack_router, "status", None),
+            description=getattr(openstack_router, "description", None),
+            project_id=getattr(openstack_router, "project_id", None),
+            is_admin_state_up=getattr(
+                openstack_router, "is_admin_state_up", None
+            ),
+            external_gateway_info=getattr(
+                openstack_router, "external_gateway_info", None
+            ),
+            is_distributed=getattr(openstack_router, "is_distributed", None),
+            is_ha=getattr(openstack_router, "is_ha", None),
+            routes=getattr(openstack_router, "routes", None),
+        )

src/openstack_mcp_server/tools/request/network.py

@@ -0,0 +1,26 @@
+from pydantic import BaseModel
+
+
+class Route(BaseModel):
+    """Static route for a router."""
+
+    destination: str
+    nexthop: str
+
+
+class ExternalFixedIP(BaseModel):
+    """External fixed IP assignment for router gateway."""
+
+    subnet_id: str | None = None
+    ip_address: str | None = None
+
+
+class ExternalGatewayInfo(BaseModel):
+    """External gateway information for a router.
+    At minimum include `network_id`. Optionally include `enable_snat` and
+    `external_fixed_ips`.
+    """
+
+    network_id: str
+    enable_snat: bool | None = None
+    external_fixed_ips: list[ExternalFixedIP] | None = None