diff --git a/litellm/proxy/common_utils/get_routes.py b/litellm/proxy/common_utils/get_routes.py new file mode 100644 index 00000000000..19465675c1a --- /dev/null +++ b/litellm/proxy/common_utils/get_routes.py @@ -0,0 +1,60 @@ +""" +Utility class for getting routes from a FastAPI app. +""" + +from typing import Any, Dict, List + +from starlette.routing import BaseRoute + + +class GetRoutes: + @staticmethod + def get_app_routes( + route: BaseRoute, + endpoint_route: Any, + ) -> List[Dict[str, Any]]: + """ + Get routes for a regular route. + """ + routes: List[Dict[str, Any]] = [] + route_info = { + "path": getattr(route, "path", None), + "methods": getattr(route, "methods", None), + "name": getattr(route, "name", None), + "endpoint": ( + endpoint_route.__name__ + if getattr(route, "endpoint", None) + else None + ), + } + routes.append(route_info) + return routes + + @staticmethod + def get_routes_for_mounted_app( + route: BaseRoute, + ) -> List[Dict[str, Any]]: + """ + Get routes for a mounted sub-application. + """ + routes: List[Dict[str, Any]] = [] + mount_path = getattr(route, 'path', '') + sub_app = getattr(route, 'app', None) + if sub_app and hasattr(sub_app, 'routes'): + for sub_route in sub_app.routes: + # Get endpoint - either from endpoint attribute or app attribute + endpoint_func = getattr(sub_route, "endpoint", None) or getattr(sub_route, "app", None) + + if endpoint_func is not None: + sub_route_path = getattr(sub_route, "path", "") + full_path = mount_path.rstrip('/') + sub_route_path + + route_info = { + "path": full_path, + "methods": getattr(sub_route, "methods", ["GET", "POST"]), + "name": getattr(sub_route, "name", None), + "endpoint": endpoint_func.__name__ if callable(endpoint_func) else None, + "mounted_app": True, + } + routes.append(route_info) + return routes \ No newline at end of file diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 9f4d4ee0f9e..15329589bb8 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -8648,21 +8648,18 @@ async def get_routes(): """ Get a list of available routes in the FastAPI application. """ + from litellm.proxy.common_utils.get_routes import GetRoutes routes = [] for route in app.routes: endpoint_route = getattr(route, "endpoint", None) if endpoint_route is not None: - route_info = { - "path": getattr(route, "path", None), - "methods": getattr(route, "methods", None), - "name": getattr(route, "name", None), - "endpoint": ( - endpoint_route.__name__ - if getattr(route, "endpoint", None) - else None - ), - } - routes.append(route_info) + routes.extend(GetRoutes.get_app_routes( + route=route, + endpoint_route=endpoint_route, + )) + # Handle mounted sub-applications (like MCP app) + elif hasattr(route, 'app') and hasattr(route, 'path'): + routes.extend(GetRoutes.get_routes_for_mounted_app(route=route)) return {"routes": routes} diff --git a/tests/test_litellm/proxy/common_utils/test_get_routes.py b/tests/test_litellm/proxy/common_utils/test_get_routes.py new file mode 100644 index 00000000000..48eadffe2e2 --- /dev/null +++ b/tests/test_litellm/proxy/common_utils/test_get_routes.py @@ -0,0 +1,168 @@ +""" +Unit tests for GetRoutes utility class. +""" + +from typing import Any, Dict, List +from unittest.mock import MagicMock, Mock + +import pytest + +from litellm.proxy.common_utils.get_routes import GetRoutes + + +class TestGetRoutes: + + def test_get_app_routes_regular_route(self): + """Test getting routes for a regular route with endpoint.""" + # Mock a regular route + mock_route = Mock() + mock_route.path = "/test/endpoint" + mock_route.methods = ["GET", "POST"] + mock_route.name = "test_endpoint" + mock_route.endpoint = Mock() + + # Mock endpoint function + mock_endpoint = Mock() + mock_endpoint.__name__ = "test_function" + + result = GetRoutes.get_app_routes(mock_route, mock_endpoint) + + assert len(result) == 1 + assert result[0]["path"] == "/test/endpoint" + assert result[0]["methods"] == ["GET", "POST"] + assert result[0]["name"] == "test_endpoint" + assert result[0]["endpoint"] == "test_function" + + def test_get_routes_for_mounted_app_regular_routes(self): + """Test getting routes for mounted app with regular API routes.""" + # Mock the main mount route + mock_mount_route = Mock() + mock_mount_route.path = "/mcp" + + # Mock sub-app with regular routes + mock_sub_app = Mock() + mock_sub_app.routes = [] + + # Create a regular API route + mock_api_route = Mock() + mock_api_route.path = "/enabled" + mock_api_route.methods = ["GET"] + mock_api_route.name = "get_mcp_server_enabled" + + # Mock endpoint function + mock_endpoint = Mock() + mock_endpoint.__name__ = "get_mcp_server_enabled" + mock_api_route.endpoint = mock_endpoint + mock_api_route.app = None # Regular route doesn't have app + + mock_sub_app.routes.append(mock_api_route) + mock_mount_route.app = mock_sub_app + + result = GetRoutes.get_routes_for_mounted_app(mock_mount_route) + + assert len(result) == 1 + assert result[0]["path"] == "/mcp/enabled" + assert result[0]["methods"] == ["GET"] + assert result[0]["name"] == "get_mcp_server_enabled" + assert result[0]["endpoint"] == "get_mcp_server_enabled" + assert result[0]["mounted_app"] is True + + def test_get_routes_for_mounted_app_mount_objects(self): + """Test getting routes for mounted app with Mount objects (the main fix).""" + # Mock the main mount route + mock_mount_route = Mock() + mock_mount_route.path = "/mcp" + + # Mock sub-app + mock_sub_app = Mock() + mock_sub_app.routes = [] + + # Create Mount object for base MCP route (path='') + mock_mount_base = Mock(spec=['path', 'name', 'endpoint', 'app']) + mock_mount_base.path = "" + mock_mount_base.name = "" + mock_mount_base.endpoint = None # Mount objects don't have endpoint + + # Mock app function + mock_app_function = Mock() + mock_app_function.__name__ = "handle_streamable_http_mcp" + mock_mount_base.app = mock_app_function + + # Create Mount object for SSE route (path='/sse') + mock_mount_sse = Mock(spec=['path', 'name', 'endpoint', 'app']) + mock_mount_sse.path = "/sse" + mock_mount_sse.name = "" + mock_mount_sse.endpoint = None # Mount objects don't have endpoint + + # Mock app function for SSE + mock_sse_function = Mock() + mock_sse_function.__name__ = "handle_sse_mcp" + mock_mount_sse.app = mock_sse_function + + mock_sub_app.routes.extend([mock_mount_base, mock_mount_sse]) + mock_mount_route.app = mock_sub_app + + result = GetRoutes.get_routes_for_mounted_app(mock_mount_route) + + # Should capture both /mcp and /mcp/sse routes + assert len(result) == 2 + + # Check base MCP route + base_route = next(r for r in result if r["path"] == "/mcp") + assert base_route["methods"] == ["GET", "POST"] # Default methods + assert base_route["endpoint"] == "handle_streamable_http_mcp" + assert base_route["mounted_app"] is True + + # Check SSE route + sse_route = next(r for r in result if r["path"] == "/mcp/sse") + assert sse_route["methods"] == ["GET", "POST"] # Default methods + assert sse_route["endpoint"] == "handle_sse_mcp" + assert sse_route["mounted_app"] is True + + def test_get_routes_for_mounted_app_mixed_routes(self): + """Test getting routes for mounted app with both regular routes and Mount objects.""" + # Mock the main mount route + mock_mount_route = Mock() + mock_mount_route.path = "/mcp" + + # Mock sub-app + mock_sub_app = Mock() + mock_sub_app.routes = [] + + # Create a regular API route + mock_api_route = Mock() + mock_api_route.path = "/enabled" + mock_api_route.methods = ["GET"] + mock_api_route.name = "get_mcp_server_enabled" + mock_endpoint = Mock() + mock_endpoint.__name__ = "get_mcp_server_enabled" + mock_api_route.endpoint = mock_endpoint + mock_api_route.app = None + + # Create Mount object + mock_mount_base = Mock(spec=['path', 'name', 'endpoint', 'app']) + mock_mount_base.path = "" + mock_mount_base.name = "" + mock_mount_base.endpoint = None + mock_app_function = Mock() + mock_app_function.__name__ = "handle_streamable_http_mcp" + mock_mount_base.app = mock_app_function + + mock_sub_app.routes.extend([mock_api_route, mock_mount_base]) + mock_mount_route.app = mock_sub_app + + result = GetRoutes.get_routes_for_mounted_app(mock_mount_route) + + # Should capture both the API route and the Mount object + assert len(result) == 2 + + # Check API route + api_route = next(r for r in result if r["path"] == "/mcp/enabled") + assert api_route["methods"] == ["GET"] + assert api_route["endpoint"] == "get_mcp_server_enabled" + + # Check Mount object route + mount_route = next(r for r in result if r["path"] == "/mcp") + assert mount_route["endpoint"] == "handle_streamable_http_mcp" + assert mount_route["mounted_app"] is True +