Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
87 changes: 87 additions & 0 deletions src/openstack_mcp_server/tools/network_tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
Network,
Port,
Router,
RouterInterface,
Subnet,
)

Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -1038,6 +1042,89 @@ 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 = {}
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),
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")
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.
Expand Down
6 changes: 6 additions & 0 deletions src/openstack_mcp_server/tools/response/network.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
65 changes: 65 additions & 0 deletions tests/tools/test_network_tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
Network,
Port,
Router,
RouterInterface,
Subnet,
)

Expand Down Expand Up @@ -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"
)