mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-07 08:26:10 +00:00
feat: initial commit for openapi to mcp conversion
allows users to give openapi spec for a server, and litellm to serve it as an mcp
This commit is contained in:
Krrish Dholakia
parent
45ec1ffdaa
commit
386468ad77
2 changed files with 249 additions and 3 deletions
|
|
@ -0,0 +1,222 @@
|
|||
"""
|
||||
This module is used to generate MCP tools from OpenAPI specs.
|
||||
"""
|
||||
|
||||
import json
|
||||
from typing import Any, Dict
|
||||
|
||||
import httpx
|
||||
|
||||
from litellm._logging import verbose_logger
|
||||
from litellm.proxy._experimental.mcp_server.tool_registry import (
|
||||
global_mcp_tool_registry,
|
||||
)
|
||||
|
||||
# Store the base URL and headers globally
|
||||
BASE_URL = ""
|
||||
HEADERS: Dict[str, str] = {}
|
||||
|
||||
|
||||
def load_openapi_spec(filepath: str) -> Dict[str, Any]:
|
||||
"""Load OpenAPI specification from JSON file."""
|
||||
with open(filepath, "r") as f:
|
||||
return json.load(f)
|
||||
|
||||
|
||||
def get_base_url(spec: Dict[str, Any]) -> str:
|
||||
"""Extract base URL from OpenAPI spec."""
|
||||
# OpenAPI 3.x
|
||||
if "servers" in spec and spec["servers"]:
|
||||
return spec["servers"][0]["url"]
|
||||
# OpenAPI 2.x (Swagger)
|
||||
elif "host" in spec:
|
||||
scheme = spec.get("schemes", ["https"])[0]
|
||||
base_path = spec.get("basePath", "")
|
||||
return f"{scheme}://{spec['host']}{base_path}"
|
||||
return ""
|
||||
|
||||
|
||||
def extract_parameters(operation: Dict[str, Any]) -> tuple:
|
||||
"""Extract parameter names from OpenAPI operation."""
|
||||
path_params = []
|
||||
query_params = []
|
||||
body_params = []
|
||||
|
||||
# OpenAPI 3.x and 2.x parameters
|
||||
if "parameters" in operation:
|
||||
for param in operation["parameters"]:
|
||||
param_name = param["name"]
|
||||
if param.get("in") == "path":
|
||||
path_params.append(param_name)
|
||||
elif param.get("in") == "query":
|
||||
query_params.append(param_name)
|
||||
elif param.get("in") == "body":
|
||||
body_params.append(param_name)
|
||||
|
||||
# OpenAPI 3.x requestBody
|
||||
if "requestBody" in operation:
|
||||
body_params.append("body")
|
||||
|
||||
return path_params, query_params, body_params
|
||||
|
||||
|
||||
def build_input_schema(operation: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Build MCP input schema from OpenAPI operation."""
|
||||
properties = {}
|
||||
required = []
|
||||
|
||||
# Process parameters
|
||||
if "parameters" in operation:
|
||||
for param in operation["parameters"]:
|
||||
param_name = param["name"]
|
||||
param_schema = param.get("schema", {})
|
||||
param_type = param_schema.get("type", "string")
|
||||
|
||||
properties[param_name] = {
|
||||
"type": param_type,
|
||||
"description": param.get("description", ""),
|
||||
}
|
||||
|
||||
if param.get("required", False):
|
||||
required.append(param_name)
|
||||
|
||||
# Process requestBody (OpenAPI 3.x)
|
||||
if "requestBody" in operation:
|
||||
request_body = operation["requestBody"]
|
||||
content = request_body.get("content", {})
|
||||
|
||||
# Try to get JSON schema
|
||||
if "application/json" in content:
|
||||
schema = content["application/json"].get("schema", {})
|
||||
properties["body"] = {
|
||||
"type": "object",
|
||||
"description": request_body.get("description", "Request body"),
|
||||
"properties": schema.get("properties", {}),
|
||||
}
|
||||
if request_body.get("required", False):
|
||||
required.append("body")
|
||||
|
||||
return {
|
||||
"type": "object",
|
||||
"properties": properties,
|
||||
"required": required if required else [],
|
||||
}
|
||||
|
||||
|
||||
def create_tool_function(
|
||||
path: str, method: str, operation: Dict[str, Any], base_url: str
|
||||
):
|
||||
"""Create a tool function for an OpenAPI operation."""
|
||||
|
||||
path_params, query_params, body_params = extract_parameters(operation)
|
||||
all_params = path_params + query_params + body_params
|
||||
|
||||
# Build function signature dynamically
|
||||
if all_params:
|
||||
params_str = ", ".join(f"{p}: str = ''" for p in all_params)
|
||||
else:
|
||||
params_str = ""
|
||||
|
||||
# Create the function code as a string
|
||||
func_code = f'''
|
||||
async def tool_function({params_str}) -> str:
|
||||
"""Dynamically generated tool function."""
|
||||
url = base_url + path
|
||||
|
||||
# Replace path parameters
|
||||
path_param_names = {path_params}
|
||||
for param_name in path_param_names:
|
||||
param_value = locals().get(param_name, "")
|
||||
if param_value:
|
||||
url = url.replace("{{" + param_name + "}}", str(param_value))
|
||||
|
||||
# Build query params
|
||||
query_param_names = {query_params}
|
||||
params = {{}}
|
||||
for param_name in query_param_names:
|
||||
param_value = locals().get(param_name, "")
|
||||
if param_value:
|
||||
params[param_name] = param_value
|
||||
|
||||
# Build request body
|
||||
body_param_names = {body_params}
|
||||
json_body = None
|
||||
if body_param_names:
|
||||
body_value = locals().get("body", {{}})
|
||||
if isinstance(body_value, dict):
|
||||
json_body = body_value
|
||||
elif body_value:
|
||||
# If it's a string, try to parse as JSON
|
||||
import json as json_module
|
||||
try:
|
||||
json_body = json_module.loads(body_value) if isinstance(body_value, str) else {{"data": body_value}}
|
||||
except:
|
||||
json_body = {{"data": body_value}}
|
||||
|
||||
# Make HTTP request
|
||||
async with httpx.AsyncClient() as client:
|
||||
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 "Unsupported HTTP method: {method}"
|
||||
|
||||
return response.text
|
||||
'''
|
||||
|
||||
# Execute the function code to create the actual function
|
||||
local_vars = {
|
||||
"httpx": httpx,
|
||||
"HEADERS": HEADERS,
|
||||
"base_url": base_url,
|
||||
"path": path,
|
||||
"method": method,
|
||||
}
|
||||
exec(func_code, local_vars)
|
||||
|
||||
return local_vars["tool_function"]
|
||||
|
||||
|
||||
def register_tools_from_openapi(spec: Dict[str, Any], base_url: str):
|
||||
"""Register MCP tools from OpenAPI specification."""
|
||||
paths = spec.get("paths", {})
|
||||
|
||||
for path, path_item in paths.items():
|
||||
for method in ["get", "post", "put", "delete", "patch"]:
|
||||
if method in path_item:
|
||||
operation = path_item[method]
|
||||
|
||||
# Generate tool name
|
||||
operation_id = operation.get(
|
||||
"operationId", f"{method}_{path.replace('/', '_')}"
|
||||
)
|
||||
tool_name = operation_id.replace(" ", "_").lower()
|
||||
|
||||
# Get description
|
||||
description = operation.get(
|
||||
"summary", operation.get("description", f"{method.upper()} {path}")
|
||||
)
|
||||
|
||||
# Build input schema
|
||||
input_schema = build_input_schema(operation)
|
||||
|
||||
# Create tool function
|
||||
tool_func = create_tool_function(path, method, operation, base_url)
|
||||
tool_func.__name__ = tool_name
|
||||
tool_func.__doc__ = description
|
||||
|
||||
# Register tool with local registry
|
||||
global_mcp_tool_registry.register_tool(
|
||||
name=tool_name,
|
||||
description=description,
|
||||
input_schema=input_schema,
|
||||
handler=tool_func,
|
||||
)
|
||||
verbose_logger.debug(f"Registered tool: {tool_name}")
|
||||
|
|
@ -1,6 +1,30 @@
|
|||
model_list:
|
||||
- model_name: gpt-5-mini
|
||||
litellm_params:
|
||||
model: azure/gpt-5-mini-2
|
||||
api_key: os.environ/AZURE_API_KEY_ALT
|
||||
api_base: os.environ/AZURE_API_BASE_ALT
|
||||
model: openai/gpt-4o-mini
|
||||
api_base: "https://webhook.site/2f385e05-00aa-402b-86d1-efc9261471a5"
|
||||
api_key: dummy
|
||||
- model_name: "byok-wildcard/*"
|
||||
litellm_params:
|
||||
model: openai/*
|
||||
- model_name: xai-grok-3
|
||||
litellm_params:
|
||||
model: xai/grok-3
|
||||
- model_name: hosted_vllm/whisper-v3
|
||||
litellm_params:
|
||||
model: hosted_vllm/whisper-v3
|
||||
api_base: "https://webhook.site/2f385e05-00aa-402b-86d1-efc9261471a5"
|
||||
api_key: dummy
|
||||
|
||||
mcp_servers:
|
||||
my_api_mcp:
|
||||
url: "http://0.0.0.0:8000"
|
||||
spec_path: "/Users/krrishdholakia/Documents/temp_py_folder/openapi.json"
|
||||
auth_type: none
|
||||
allowed_tools: ["get_users", "create_user"]
|
||||
access_groups: ["dev_group", "api_team"]
|
||||
|
||||
|
||||
litellm_settings:
|
||||
callbacks: ["prometheus"]
|
||||
custom_prometheus_metadata_labels: ["metadata.initiative", "metadata.business-unit"]
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue