diff --git a/litellm/proxy/_experimental/mcp_server/openapi_to_mcp_generator.py b/litellm/proxy/_experimental/mcp_server/openapi_to_mcp_generator.py index bbbb8ad4a3..8cdb381cf3 100644 --- a/litellm/proxy/_experimental/mcp_server/openapi_to_mcp_generator.py +++ b/litellm/proxy/_experimental/mcp_server/openapi_to_mcp_generator.py @@ -7,12 +7,12 @@ from pathlib import PurePosixPath from typing import Any, Dict, Optional from urllib.parse import quote +import httpx + from litellm._logging import verbose_logger -from litellm.llms.custom_httpx.http_handler import get_async_httpx_client from litellm.proxy._experimental.mcp_server.tool_registry import ( global_mcp_tool_registry, ) -from litellm.types.llms.custom_http import httpxSpecialProvider # Store the base URL and headers globally BASE_URL = "" @@ -42,79 +42,6 @@ def _sanitize_path_parameter_value(param_value: Any, param_name: str) -> str: return quote(value_str, safe="") -async def _invoke_openapi_operation( - *, - method: str, - base_url: str, - path: str, - headers: Dict[str, str], - path_params: list, - query_params: list, - body_params: list, - provided_params: Dict[str, Any], -) -> str: - """Execute the OpenAPI operation using provided parameters.""" - - url = base_url + path - - # Replace path parameters with sanitized values - for param_name in path_params: - param_value = provided_params.get(param_name, "") - if param_value: - try: - safe_value = _sanitize_path_parameter_value(param_value, param_name) - except ValueError as exc: - return "Invalid path parameter: " + str(exc) - url = url.replace("{" + param_name + "}", safe_value) - - # Build query params - params: Dict[str, Any] = {} - for param_name in query_params: - param_value = provided_params.get(param_name, "") - if param_value: - params[param_name] = param_value - - # Build request body - json_body = None - if body_params: - body_value = provided_params.get("body", {}) - if isinstance(body_value, dict): - json_body = body_value - elif body_value: - import json as json_module - - try: - json_body = ( - json_module.loads(body_value) - if isinstance(body_value, str) - else {"data": body_value} - ) - except Exception: - json_body = {"data": body_value} - - client = get_async_httpx_client(llm_provider=httpxSpecialProvider.MCP) - - method_lower = method.lower() - if method_lower == "get": - response = await client.get(url, params=params, headers=headers) - elif method_lower == "post": - response = await client.post( - url, params=params, json=json_body, headers=headers - ) - elif method_lower == "put": - response = await client.put(url, params=params, json=json_body, headers=headers) - elif method_lower == "delete": - response = await client.delete(url, params=params, headers=headers) - elif method_lower == "patch": - response = await client.patch( - url, params=params, json=json_body, headers=headers - ) - else: - return f"Unsupported HTTP method: {method}" - - return response.text - - def load_openapi_spec(filepath: str) -> Dict[str, Any]: """Load OpenAPI specification from JSON file.""" with open(filepath, "r") as f: @@ -210,55 +137,107 @@ def create_tool_function( ): """Create a tool function for an OpenAPI operation. + This function creates an async tool function that can be called with + keyword arguments. Parameter names from the OpenAPI spec are accessed + directly via **kwargs, avoiding syntax errors from invalid Python identifiers. + Args: path: API endpoint path method: HTTP method (get, post, put, delete, patch) operation: OpenAPI operation object base_url: Base URL for the API headers: Optional headers to include in requests (e.g., authentication) + + Returns: + An async function that accepts **kwargs and makes the HTTP request """ if headers is None: headers = {} path_params, query_params, body_params = extract_parameters(operation) - all_params = path_params + query_params + body_params + original_method = method.lower() - # Build function signature dynamically - if all_params: - params_str = ", ".join(f"{p}: str = ''" for p in all_params) - else: - params_str = "" + async def tool_function(**kwargs: Any) -> str: + """ + Dynamically generated tool function. - # Create the function code as a string - func_code = f''' -async def tool_function({params_str}) -> str: - """Dynamically generated tool function.""" - provided_params = {{}} - for param_name in {all_params}: - provided_params[param_name] = locals().get(param_name, "") - return await _invoke_openapi_operation( - method="{method}", - base_url=base_url, - path=path, - headers=headers, - path_params={path_params}, - query_params={query_params}, - body_params={body_params}, - provided_params=provided_params, - ) -''' + Accepts keyword arguments where keys are the original OpenAPI parameter names. + The function safely handles parameter names that aren't valid Python identifiers + by using **kwargs instead of named parameters. + """ + # Build URL from base_url and path + url = base_url + path - # Execute the function code to create the actual function - local_vars = { - "headers": headers, - "base_url": base_url, - "path": path, - "method": method, - "_invoke_openapi_operation": _invoke_openapi_operation, - } - exec(func_code, local_vars) + # Replace path parameters using original names from OpenAPI spec + # Apply path traversal validation and URL encoding + for param_name in path_params: + param_value = kwargs.get(param_name, "") + if param_value: + try: + # Sanitize and encode path parameter to prevent traversal attacks + safe_value = _sanitize_path_parameter_value(param_value, param_name) + except ValueError as exc: + return "Invalid path parameter: " + str(exc) + # Replace {param_name} or {{param_name}} in URL + url = url.replace("{" + param_name + "}", safe_value) + url = url.replace("{{" + param_name + "}}", safe_value) - return local_vars["tool_function"] + # Build query params using original parameter names + params: Dict[str, Any] = {} + for param_name in query_params: + param_value = kwargs.get(param_name, "") + if param_value: + # Use original parameter name in query string (as expected by API) + params[param_name] = param_value + + # Build request body + json_body: Optional[Dict[str, Any]] = None + if body_params: + # Try "body" first (most common), then check all body param names + body_value = kwargs.get("body", {}) + if not body_value: + for param_name in body_params: + body_value = kwargs.get(param_name, {}) + if body_value: + break + + if isinstance(body_value, dict): + json_body = body_value + elif body_value: + # If it's a string, try to parse as JSON + try: + json_body = ( + json.loads(body_value) + if isinstance(body_value, str) + else {"data": body_value} + ) + except (json.JSONDecodeError, TypeError): + json_body = {"data": body_value} + + # Make HTTP request + async with httpx.AsyncClient() as client: + if original_method == "get": + response = await client.get(url, params=params, headers=headers) + elif original_method == "post": + response = await client.post( + url, params=params, json=json_body, headers=headers + ) + elif original_method == "put": + response = await client.put( + url, params=params, json=json_body, headers=headers + ) + elif original_method == "delete": + response = await client.delete(url, params=params, headers=headers) + elif original_method == "patch": + response = await client.patch( + url, params=params, json=json_body, headers=headers + ) + else: + return f"Unsupported HTTP method: {original_method}" + + return response.text + + return tool_function def register_tools_from_openapi(spec: Dict[str, Any], base_url: str): diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/test_openapi_to_mcp_generator.py b/tests/test_litellm/proxy/_experimental/mcp_server/test_openapi_to_mcp_generator.py index 71668d4af4..1044ad3ed2 100644 --- a/tests/test_litellm/proxy/_experimental/mcp_server/test_openapi_to_mcp_generator.py +++ b/tests/test_litellm/proxy/_experimental/mcp_server/test_openapi_to_mcp_generator.py @@ -1,101 +1,522 @@ -"""Tests for OpenAPI to MCP generator path handling.""" +""" +Tests for OpenAPI to MCP generator, focusing on security and edge cases. + +This test suite ensures that: +1. Parameter names with invalid Python identifiers are handled safely +2. No exec() is used (security) +3. All edge cases (hyphens, dots, keywords, special chars) work correctly +4. Path traversal attacks are prevented +5. Path parameters are properly URL encoded +""" import pytest +from unittest.mock import AsyncMock, patch from litellm.proxy._experimental.mcp_server.openapi_to_mcp_generator import ( create_tool_function, + build_input_schema, + extract_parameters, ) -class _DummyResponse: - def __init__(self, text: str = "ok"): - self.text = text +class TestCreateToolFunction: + """Test create_tool_function with various parameter name edge cases.""" + @pytest.mark.asyncio + async def test_hyphenated_path_parameter(self): + """Test function with hyphenated path parameter (e.g., repository-id).""" + operation = { + "parameters": [ + { + "name": "repository-id", + "in": "path", + "required": True, + "schema": {"type": "string"}, + } + ] + } -class _DummyAsyncClient: - """Minimal async client stub that records requests.""" + func = create_tool_function( + path="/repos/{repository-id}", + method="get", + operation=operation, + base_url="https://api.example.com", + ) - last_instance = None + # Should not raise SyntaxError + assert callable(func) + assert func.__name__ == "tool_function" - def __init__(self): - self.requests = [] - _DummyAsyncClient.last_instance = self + # Test calling with original parameter name + with patch("httpx.AsyncClient") as mock_client: + mock_response = AsyncMock() + mock_response.text = '{"id": "123"}' + mock_client.return_value.__aenter__.return_value.get = AsyncMock( + return_value=mock_response + ) - async def __aenter__(self): - return self + result = await func(**{"repository-id": "test-repo"}) + assert result == '{"id": "123"}' - async def __aexit__(self, exc_type, exc, tb): - return False + # Verify URL was constructed correctly + call_args = mock_client.return_value.__aenter__.return_value.get.call_args + assert "repository-id" in str(call_args[0][0]) or "test-repo" in str( + call_args[0][0] + ) - async def get(self, url, params=None, headers=None): - self.requests.append(("get", url, params, headers)) - return _DummyResponse("dummy-response") + @pytest.mark.asyncio + async def test_leading_digit_parameter(self): + """Test function with parameter starting with digit (e.g., 2fa-code).""" + operation = { + "parameters": [ + { + "name": "2fa-code", + "in": "query", + "required": False, + "schema": {"type": "string"}, + } + ] + } + func = create_tool_function( + path="/verify", + method="post", + operation=operation, + base_url="https://api.example.com", + ) -@pytest.mark.asyncio -async def test_should_reject_path_traversal_inputs(monkeypatch): - monkeypatch.setattr( - "litellm.proxy._experimental.mcp_server.openapi_to_mcp_generator.get_async_httpx_client", - lambda *_, **__: _DummyAsyncClient(), - ) - _DummyAsyncClient.last_instance = None + assert callable(func) - operation = { - "parameters": [ - { - "name": "filename", - "in": "path", + with patch("httpx.AsyncClient") as mock_client: + mock_response = AsyncMock() + mock_response.text = "verified" + mock_client.return_value.__aenter__.return_value.post = AsyncMock( + return_value=mock_response + ) + + result = await func(**{"2fa-code": "123456"}) + assert result == "verified" + + # Verify query parameter was included + call_args = mock_client.return_value.__aenter__.return_value.post.call_args + assert call_args[1]["params"]["2fa-code"] == "123456" + + @pytest.mark.asyncio + async def test_dot_in_parameter_name(self): + """Test function with dot in parameter name (e.g., user.name).""" + operation = { + "parameters": [ + { + "name": "user.name", + "in": "query", + "required": False, + "schema": {"type": "string"}, + } + ] + } + + func = create_tool_function( + path="/search", + method="get", + operation=operation, + base_url="https://api.example.com", + ) + + assert callable(func) + + with patch("httpx.AsyncClient") as mock_client: + mock_response = AsyncMock() + mock_response.text = "found" + mock_client.return_value.__aenter__.return_value.get = AsyncMock( + return_value=mock_response + ) + + result = await func(**{"user.name": "john.doe"}) + assert result == "found" + + call_args = mock_client.return_value.__aenter__.return_value.get.call_args + assert call_args[1]["params"]["user.name"] == "john.doe" + + @pytest.mark.asyncio + async def test_dollar_sign_parameter(self): + """Test function with dollar sign parameter (OData style, e.g., $filter).""" + operation = { + "parameters": [ + { + "name": "$filter", + "in": "query", + "required": False, + "schema": {"type": "string"}, + } + ] + } + + func = create_tool_function( + path="/entities", + method="get", + operation=operation, + base_url="https://api.example.com", + ) + + assert callable(func) + + with patch("httpx.AsyncClient") as mock_client: + mock_response = AsyncMock() + mock_response.text = "[]" + mock_client.return_value.__aenter__.return_value.get = AsyncMock( + return_value=mock_response + ) + + result = await func(**{"$filter": "name eq 'test'"}) + assert result == "[]" + + call_args = mock_client.return_value.__aenter__.return_value.get.call_args + assert call_args[1]["params"]["$filter"] == "name eq 'test'" + + @pytest.mark.asyncio + async def test_python_keyword_parameter(self): + """Test function with Python keyword as parameter name (e.g., class).""" + operation = { + "parameters": [ + { + "name": "class", + "in": "query", + "required": False, + "schema": {"type": "string"}, + } + ] + } + + func = create_tool_function( + path="/items", + method="get", + operation=operation, + base_url="https://api.example.com", + ) + + assert callable(func) + + with patch("httpx.AsyncClient") as mock_client: + mock_response = AsyncMock() + mock_response.text = "items" + mock_client.return_value.__aenter__.return_value.get = AsyncMock( + return_value=mock_response + ) + + result = await func(**{"class": "premium"}) + assert result == "items" + + call_args = mock_client.return_value.__aenter__.return_value.get.call_args + assert call_args[1]["params"]["class"] == "premium" + + @pytest.mark.asyncio + async def test_multiple_problematic_parameters(self): + """Test function with multiple problematic parameter names.""" + operation = { + "parameters": [ + { + "name": "repository-id", + "in": "path", + "required": True, + "schema": {"type": "string"}, + }, + { + "name": "2fa-code", + "in": "query", + "required": False, + "schema": {"type": "string"}, + }, + { + "name": "$filter", + "in": "query", + "required": False, + "schema": {"type": "string"}, + }, + ] + } + + func = create_tool_function( + path="/repos/{repository-id}", + method="get", + operation=operation, + base_url="https://api.example.com", + ) + + assert callable(func) + + with patch("httpx.AsyncClient") as mock_client: + mock_response = AsyncMock() + mock_response.text = "success" + mock_client.return_value.__aenter__.return_value.get = AsyncMock( + return_value=mock_response + ) + + result = await func( + **{ + "repository-id": "test-repo", + "2fa-code": "123", + "$filter": "active", + } + ) + assert result == "success" + + @pytest.mark.asyncio + async def test_request_body_parameter(self): + """Test function with request body parameter.""" + operation = { + "requestBody": { "required": True, - "schema": {"type": "string"}, + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": {"name": {"type": "string"}}, + } + } + }, } - ] - } + } - tool_function = create_tool_function( - path="/files/{filename}", - method="GET", - operation=operation, - base_url="https://example.com", - ) + func = create_tool_function( + path="/create", + method="post", + operation=operation, + base_url="https://api.example.com", + ) - response = await tool_function(filename="../admin") + assert callable(func) - assert "Invalid path parameter" in response + with patch("httpx.AsyncClient") as mock_client: + mock_response = AsyncMock() + mock_response.text = "created" + mock_client.return_value.__aenter__.return_value.post = AsyncMock( + return_value=mock_response + ) + result = await func(**{"body": {"name": "test"}}) + assert result == "created" -@pytest.mark.asyncio -async def test_should_encode_and_request_safe_path_parameters(monkeypatch): - monkeypatch.setattr( - "litellm.proxy._experimental.mcp_server.openapi_to_mcp_generator.get_async_httpx_client", - lambda *_, **__: _DummyAsyncClient(), - ) - _DummyAsyncClient.last_instance = None + call_args = mock_client.return_value.__aenter__.return_value.post.call_args + assert call_args[1]["json"] == {"name": "test"} - operation = { - "parameters": [ - { - "name": "filename", - "in": "path", - "required": True, - "schema": {"type": "string"}, + @pytest.mark.asyncio + async def test_no_parameters(self): + """Test function with no parameters.""" + operation = {} + + func = create_tool_function( + path="/health", + method="get", + operation=operation, + base_url="https://api.example.com", + ) + + assert callable(func) + + with patch("httpx.AsyncClient") as mock_client: + mock_response = AsyncMock() + mock_response.text = "ok" + mock_client.return_value.__aenter__.return_value.get = AsyncMock( + return_value=mock_response + ) + + result = await func() + assert result == "ok" + + @pytest.mark.asyncio + async def test_all_http_methods(self): + """Test all supported HTTP methods.""" + methods = ["get", "post", "put", "delete", "patch"] + + for method in methods: + operation = { + "parameters": [ + { + "name": "repository-id", + "in": "path", + "required": True, + "schema": {"type": "string"}, + } + ] } - ] - } - tool_function = create_tool_function( - path="/files/{filename}", - method="GET", - operation=operation, - base_url="https://example.com", - ) + func = create_tool_function( + path="/repos/{repository-id}", + method=method, + operation=operation, + base_url="https://api.example.com", + ) - response = await tool_function(filename="report 2024.json") + assert callable(func) - assert response == "dummy-response" + with patch("httpx.AsyncClient") as mock_client: + mock_response = AsyncMock() + mock_response.text = "success" - dummy_client = _DummyAsyncClient.last_instance - assert dummy_client is not None - method, url, params, headers = dummy_client.requests[0] - assert method == "get" - assert url == "https://example.com/files/report%202024.json" - assert params == {} + client_method = getattr( + mock_client.return_value.__aenter__.return_value, method + ) + client_method.return_value = mock_response + client_method = AsyncMock(return_value=mock_response) + setattr( + mock_client.return_value.__aenter__.return_value, + method, + client_method, + ) + + result = await func(**{"repository-id": "test"}) + assert result == "success" + + def test_no_exec_usage(self): + """Verify that create_tool_function does not use exec().""" + import ast + import inspect + + # Get the source code of create_tool_function + source = inspect.getsource(create_tool_function) + + # Parse the AST + tree = ast.parse(source) + + # Check for exec() calls + exec_calls = [] + for node in ast.walk(tree): + if isinstance(node, ast.Call): + if isinstance(node.func, ast.Name) and node.func.id == "exec": + exec_calls.append(node) + + # Should have no exec() calls + assert len(exec_calls) == 0, "create_tool_function should not use exec()" + + +class TestBuildInputSchema: + """Test that build_input_schema preserves original parameter names.""" + + def test_original_parameter_names_preserved(self): + """Test that original parameter names are preserved in input schema.""" + operation = { + "parameters": [ + { + "name": "repository-id", + "in": "path", + "required": True, + "schema": {"type": "string"}, + }, + { + "name": "2fa-code", + "in": "query", + "required": False, + "schema": {"type": "string"}, + }, + { + "name": "$filter", + "in": "query", + "required": False, + "schema": {"type": "string"}, + }, + ] + } + + schema = build_input_schema(operation) + + # Original names should be in the schema + assert "repository-id" in schema["properties"] + assert "2fa-code" in schema["properties"] + assert "$filter" in schema["properties"] + + # Required should include original names + assert "repository-id" in schema["required"] + + +class TestExtractParameters: + """Test parameter extraction from OpenAPI operations.""" + + def test_extract_path_query_body_params(self): + """Test extraction of different parameter types.""" + operation = { + "parameters": [ + {"name": "repo-id", "in": "path"}, + {"name": "filter", "in": "query"}, + {"name": "data", "in": "body"}, + ], + "requestBody": { + "content": {"application/json": {"schema": {"type": "object"}}} + }, + } + + path_params, query_params, body_params = extract_parameters(operation) + + assert "repo-id" in path_params + assert "filter" in query_params + assert "data" in body_params + assert "body" in body_params # From requestBody + + +if __name__ == "__main__": + pytest.main([__file__, "-v"]) + + +class TestPathSecurity: + """Test path traversal security and URL encoding.""" + + @pytest.mark.asyncio + async def test_should_reject_path_traversal_inputs(self): + """Test that path traversal attacks (../admin) are rejected.""" + operation = { + "parameters": [ + { + "name": "filename", + "in": "path", + "required": True, + "schema": {"type": "string"}, + } + ] + } + + tool_function = create_tool_function( + path="/files/{filename}", + method="GET", + operation=operation, + base_url="https://example.com", + ) + + response = await tool_function(**{"filename": "../admin"}) + + assert "Invalid path parameter" in response + + @pytest.mark.asyncio + async def test_should_encode_and_request_safe_path_parameters(self): + """Test that path parameters are properly URL encoded.""" + operation = { + "parameters": [ + { + "name": "filename", + "in": "path", + "required": True, + "schema": {"type": "string"}, + } + ] + } + + tool_function = create_tool_function( + path="/files/{filename}", + method="GET", + operation=operation, + base_url="https://example.com", + ) + + with patch("httpx.AsyncClient") as mock_client: + mock_response = AsyncMock() + mock_response.text = "dummy-response" + mock_client.return_value.__aenter__.return_value.get = AsyncMock( + return_value=mock_response + ) + + response = await tool_function(**{"filename": "report 2024.json"}) + + assert response == "dummy-response" + + # Verify URL was properly encoded + call_args = mock_client.return_value.__aenter__.return_value.get.call_args + url = call_args[0][0] + assert url == "https://example.com/files/report%202024.json"