Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
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
91 changes: 91 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,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
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

제가 보기엔 다 훌륭하신데 궁금해서 댓글 드립니다.
_proxy.py를 확인해보니 subnet_id와 port_id를 전달하지 않아도 None으로 처리하던데, not None 조건문을 태우는게 더 우아한방식일까요?

Copy link
Collaborator Author

@platanus-kr platanus-kr Oct 11, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

프록시 전달 목적의 argument를 만들기 위해 처리되는 로직입니다.
tools는 string으로 받지만 프록시 인터페이스는 dict로 받고 있으니, 형변환 과정이라고 봐주시면 되겠습니다.


(보충설명)

조금 이야기를 보태자면, 라우터에 인터페이스를 추가하기 위해서는 port나 subnet 중 무조건 둘 중에 하나를 지정해야 합니다.
서브넷을 지정하는 경우 해당 서브넷에서 포트를 생성해서 라우터에 붙이는 방식이고,
포트를 지정하는 경우는 별도 과정 없이 라우터에 붙입니다.

즉, 확인하신 로직은 포트와 서브넷이 바디에 있을때, 어느것을 우선으로 지정해서 보낼건지 결정하는 부분입니다.

    def add_interface_to_router(self, router, subnet_id=None, port_id=None):
        body = {}
        if port_id:
            body = {'port_id': port_id}
        else:
            body = {'subnet_id': subnet_id}
        router = self._get_resource(_router.Router, router)
        return router.add_interface(self, **body)

Copy link
Collaborator

@jja6312 jja6312 Oct 11, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

서브넷 지정시 포트가 생성되는지 몰랐네요~ 보충설명까지 감사합니다!!
근데 제가 질문이 구체적이지 않아 의도 전달이 불충분하지 않았나 한데요

현재 코드에서

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: dict = {}
            args["subnet_id"] = subnet_id
            args["port_id"] = port_id

로 작성한다면 openstacksdk/.../_proxy.py의 add_interface_to_router 함수가 실행될 때 해당 함수가 subnet_id, port_id 매개변수를 기본값 None으로 설정하기때문에 두 코드 다 결론적으로는 같은 동작을 할 것으로 보이는데,
그래도 if문을 통해 철옹성 같은?ㅎㅎ 안전한 코드를 짜는게 좋은가 궁금하여 질문드렸습니다

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

이부분은 수정해서 subnet, port 둘 다 테스트해보니 잘 되네요.
반영하도록 하겠습니다!

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
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
subnet_id = None

반복문 시작 시 subnet_id=None으로 초기화하므로, else 블록의 초기화 로직은 제거해도 될 것 같습니다.

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

넵! 반영되었습니다 ㅎㅎ

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