From f5f3dcf6e8ceea497cf5087b5bcd8c3bd71bb802 Mon Sep 17 00:00:00 2001 From: platanus-kr Date: Sat, 11 Oct 2025 16:04:10 +0900 Subject: [PATCH 1/3] feat(network): add router interface tools (#82) --- .../tools/network_tools.py | 91 +++++++++++++++++++ .../tools/response/network.py | 6 ++ tests/tools/test_network_tools.py | 65 +++++++++++++ 3 files changed, 162 insertions(+) diff --git a/src/openstack_mcp_server/tools/network_tools.py b/src/openstack_mcp_server/tools/network_tools.py index 5f67469..10fcdb4 100644 --- a/src/openstack_mcp_server/tools/network_tools.py +++ b/src/openstack_mcp_server/tools/network_tools.py @@ -10,6 +10,7 @@ Network, Port, Router, + RouterInterface, Subnet, ) @@ -52,6 +53,9 @@ def register_tools(self, mcp: FastMCP): mcp.tool()(self.get_router_detail) mcp.tool()(self.update_router) mcp.tool()(self.delete_router) + mcp.tool()(self.add_router_interface) + mcp.tool()(self.get_router_interfaces) + mcp.tool()(self.remove_router_interface) def get_networks( self, @@ -1038,6 +1042,93 @@ def delete_router(self, router_id: str) -> None: conn.network.delete_router(router_id, ignore_missing=False) return None + def add_router_interface( + self, + router_id: str, + subnet_id: str | None = None, + port_id: str | None = None, + ) -> RouterInterface: + """ + Add an interface to a Router by subnet or port. + Provide either subnet_id or port_id. + + :param router_id: Target router ID + :param subnet_id: Subnet ID to attach (mutually exclusive with port_id) + :param port_id: Port ID to attach (mutually exclusive with subnet_id) + :return: Created/attached router interface information as RouterInterface + """ + conn = get_openstack_conn() + args: dict = {} + if subnet_id is not None: + args["subnet_id"] = subnet_id + if port_id is not None: + args["port_id"] = port_id + res = conn.network.add_interface_to_router(router_id, **args) + return RouterInterface( + router_id=res.get("router_id", router_id), + port_id=res.get("port_id"), + subnet_id=res.get("subnet_id"), + ) + + def get_router_interfaces(self, router_id: str) -> list[RouterInterface]: + """ + List interfaces attached to a Router. + + :param router_id: Target router ID + :return: List of RouterInterface objects representing router-owned ports + """ + conn = get_openstack_conn() + filters = { + "device_id": router_id, + "device_owner": "network:router_interface", + } + ports = conn.network.ports(**filters) + result: list[RouterInterface] = [] + for p in ports: + subnet_id = None + if getattr(p, "fixed_ips", None): + first = p.fixed_ips[0] + if isinstance(first, dict): + subnet_id = first.get("subnet_id") + else: + subnet_id = None + result.append( + RouterInterface( + router_id=router_id, + port_id=p.id, + subnet_id=subnet_id, + ) + ) + return result + + def remove_router_interface( + self, + router_id: str, + subnet_id: str | None = None, + port_id: str | None = None, + ) -> RouterInterface: + """ + Remove an interface from a Router by subnet or port. + Provide either subnet_id or port_id. + + :param router_id: Target router ID + :param subnet_id: Subnet ID to detach (mutually exclusive with port_id) + :param port_id: Port ID to detach (mutually exclusive with subnet_id) + :return: Detached interface information as RouterInterface + """ + conn = get_openstack_conn() + args: dict = {} + if subnet_id is not None: + args["subnet_id"] = subnet_id + if port_id is not None: + args["port_id"] = port_id + res = conn.network.remove_interface_from_router(router_id, **args) + return RouterInterface( + router_id=res.get("router_id", router_id), + port_id=res.get("port_id"), + subnet_id=res.get("subnet_id"), + ) + def _convert_to_router_model(self, openstack_router) -> Router: """ Convert an OpenStack Router object to a Router pydantic model. diff --git a/src/openstack_mcp_server/tools/response/network.py b/src/openstack_mcp_server/tools/response/network.py index 6894bd5..b19ab56 100644 --- a/src/openstack_mcp_server/tools/response/network.py +++ b/src/openstack_mcp_server/tools/response/network.py @@ -59,6 +59,12 @@ class Router(BaseModel): routes: list[dict] | None = None +class RouterInterface(BaseModel): + router_id: str + port_id: str + subnet_id: str | None = None + + class SecurityGroup(BaseModel): id: str name: str | None = None diff --git a/tests/tools/test_network_tools.py b/tests/tools/test_network_tools.py index 3b06996..c85655b 100644 --- a/tests/tools/test_network_tools.py +++ b/tests/tools/test_network_tools.py @@ -10,6 +10,7 @@ Network, Port, Router, + RouterInterface, Subnet, ) @@ -1569,3 +1570,67 @@ def test_delete_router_success(self, mock_openstack_connect_network): "router-6", ignore_missing=False, ) + + def test_add_get_remove_router_interface_by_subnet( + self, mock_openstack_connect_network + ): + mock_conn = mock_openstack_connect_network + + add_res = {"router_id": "r-if-1", "port_id": "p-1", "subnet_id": "s-1"} + mock_conn.network.add_interface_to_router.return_value = add_res + + p = Mock() + p.id = "p-1" + p.fixed_ips = [{"subnet_id": "s-1", "ip_address": "10.0.0.1"}] + mock_conn.network.ports.return_value = [p] + + rm_res = {"router_id": "r-if-1", "port_id": "p-1", "subnet_id": "s-1"} + mock_conn.network.remove_interface_from_router.return_value = rm_res + + tools = self.get_network_tools() + added = tools.add_router_interface("r-if-1", subnet_id="s-1") + assert added == RouterInterface( + router_id="r-if-1", port_id="p-1", subnet_id="s-1" + ) + + lst = tools.get_router_interfaces("r-if-1") + assert lst == [ + RouterInterface(router_id="r-if-1", port_id="p-1", subnet_id="s-1") + ] + + removed = tools.remove_router_interface("r-if-1", subnet_id="s-1") + assert removed == RouterInterface( + router_id="r-if-1", port_id="p-1", subnet_id="s-1" + ) + + def test_add_get_remove_router_interface_by_port( + self, mock_openstack_connect_network + ): + mock_conn = mock_openstack_connect_network + + add_res = {"router_id": "r-if-2", "port_id": "p-2", "subnet_id": "s-2"} + mock_conn.network.add_interface_to_router.return_value = add_res + + p = Mock() + p.id = "p-2" + p.fixed_ips = [{"subnet_id": "s-2", "ip_address": "10.0.1.1"}] + mock_conn.network.ports.return_value = [p] + + rm_res = {"router_id": "r-if-2", "port_id": "p-2", "subnet_id": "s-2"} + mock_conn.network.remove_interface_from_router.return_value = rm_res + + tools = self.get_network_tools() + added = tools.add_router_interface("r-if-2", port_id="p-2") + assert added == RouterInterface( + router_id="r-if-2", port_id="p-2", subnet_id="s-2" + ) + + lst = tools.get_router_interfaces("r-if-2") + assert lst == [ + RouterInterface(router_id="r-if-2", port_id="p-2", subnet_id="s-2") + ] + + removed = tools.remove_router_interface("r-if-2", port_id="p-2") + assert removed == RouterInterface( + router_id="r-if-2", port_id="p-2", subnet_id="s-2" + ) From a62d839dfa400314c2dde7446a4d367bda54fcc8 Mon Sep 17 00:00:00 2001 From: platanus-kr Date: Mon, 13 Oct 2025 20:15:34 +0900 Subject: [PATCH 2/3] improve(network): simplify if condition to adding router interface (#82) --- src/openstack_mcp_server/tools/network_tools.py | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/src/openstack_mcp_server/tools/network_tools.py b/src/openstack_mcp_server/tools/network_tools.py index 10fcdb4..5070819 100644 --- a/src/openstack_mcp_server/tools/network_tools.py +++ b/src/openstack_mcp_server/tools/network_tools.py @@ -1059,10 +1059,8 @@ def add_router_interface( """ conn = get_openstack_conn() args: dict = {} - if subnet_id is not None: - args["subnet_id"] = subnet_id - if port_id is not None: - args["port_id"] = port_id + args["subnet_id"] = subnet_id + args["port_id"] = port_id res = conn.network.add_interface_to_router(router_id, **args) return RouterInterface( router_id=res.get("router_id", router_id), From b84b7d76874b563dcfd54cc2545743ee0ac31050 Mon Sep 17 00:00:00 2001 From: platanus-kr Date: Mon, 13 Oct 2025 20:20:02 +0900 Subject: [PATCH 3/3] improve(network): simplify if condition to retrieving router interface (#82) --- src/openstack_mcp_server/tools/network_tools.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/src/openstack_mcp_server/tools/network_tools.py b/src/openstack_mcp_server/tools/network_tools.py index 5070819..62e5d55 100644 --- a/src/openstack_mcp_server/tools/network_tools.py +++ b/src/openstack_mcp_server/tools/network_tools.py @@ -1088,8 +1088,6 @@ def get_router_interfaces(self, router_id: str) -> list[RouterInterface]: first = p.fixed_ips[0] if isinstance(first, dict): subnet_id = first.get("subnet_id") - else: - subnet_id = None result.append( RouterInterface( router_id=router_id,