This is page 2 of 2. Use http://codebase.md/matthewhand/mcp-openapi-proxy?page={x} to view the full context.
# Directory Structure
```
├── .flake8
├── .github
│ └── workflows
│ ├── python-pytest.yml
│ └── testpypi.yaml
├── .gitignore
├── examples
│ ├── apis.guru-claude_desktop_config.json
│ ├── asana-claude_desktop_config.json
│ ├── box-claude_desktop_config.json
│ ├── elevenlabs-claude_desktop_config.json
│ ├── flyio-claude_desktop_config.json
│ ├── getzep-claude_desktop_config.json
│ ├── getzep.swagger.json
│ ├── glama-claude_desktop_config.json
│ ├── netbox-claude_desktop_config.json
│ ├── notion-claude_desktop_config.json
│ ├── render-claude_desktop_config.json
│ ├── slack-claude_desktop_config.json
│ ├── virustotal-claude_desktop_config.json
│ ├── virustotal.openapi.yml
│ ├── WIP-jellyfin-claude_desktop_config.json
│ └── wolframalpha-claude_desktop_config.json
├── LICENSE
├── mcp_openapi_proxy
│ ├── __init__.py
│ ├── handlers.py
│ ├── logging_setup.py
│ ├── openapi.py
│ ├── server_fastmcp.py
│ ├── server_lowlevel.py
│ ├── types.py
│ └── utils.py
├── pyproject.toml
├── README.md
├── sample_mcpServers.json
├── scripts
│ └── diagnose_examples.py
├── tests
│ ├── conftest.py
│ ├── fixtures
│ │ └── sample_openapi_specs
│ │ └── petstore_openapi_v3.json
│ ├── integration
│ │ ├── test_apisguru_integration.py
│ │ ├── test_asana_integration.py
│ │ ├── test_box_integration.py
│ │ ├── test_elevenlabs_integration.py
│ │ ├── test_example_configs.py
│ │ ├── test_fly_machines_integration.py
│ │ ├── test_getzep_integration.py
│ │ ├── test_integration_json_access.py
│ │ ├── test_jellyfin_public_demo.py
│ │ ├── test_netbox_integration.py
│ │ ├── test_notion_integration.py
│ │ ├── test_openapi_integration.py
│ │ ├── test_openwebui_integration.py
│ │ ├── test_petstore_api_existence.py
│ │ ├── test_render_integration_lowlevel.py
│ │ ├── test_render_integration.py
│ │ ├── test_slack_integration.py
│ │ ├── test_ssl_verification.py
│ │ ├── test_tool_invocation.py
│ │ ├── test_tool_prefix.py
│ │ ├── test_virustotal_integration.py
│ │ └── test_wolframalpha_integration.py
│ └── unit
│ ├── test_additional_headers.py
│ ├── test_capabilities.py
│ ├── test_embedded_openapi_json.py
│ ├── test_input_schema_generation.py
│ ├── test_mcp_tools.py
│ ├── test_openapi_spec_parser.py
│ ├── test_openapi_tool_name_length.py
│ ├── test_openapi.py
│ ├── test_parameter_substitution.py
│ ├── test_prompts.py
│ ├── test_resources.py
│ ├── test_tool_whitelisting.py
│ ├── test_uri_substitution.py
│ ├── test_utils_whitelist.py
│ └── test_utils.py
├── upload_readme_to_readme.py
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/tests/integration/test_box_integration.py:
--------------------------------------------------------------------------------
```python
"""
Integration tests for Box API via mcp-openapi-proxy, FastMCP mode.
Requires BOX_API_KEY in .env to run.
"""
import os
import json
import pytest
from dotenv import load_dotenv
from mcp_openapi_proxy.utils import fetch_openapi_spec
from mcp_openapi_proxy.server_fastmcp import list_functions, call_function
# Load .env file from project root if it exists
load_dotenv(dotenv_path=os.path.join(os.path.dirname(__file__), '../../.env'))
# --- Configuration ---
BOX_API_KEY = os.getenv("BOX_API_KEY")
# Use the spec from APIs.guru directory
SPEC_URL = "https://raw.githubusercontent.com/APIs-guru/openapi-directory/main/APIs/box.com/2.0.0/openapi.yaml"
# Whitelist the endpoints needed for these tests
TOOL_WHITELIST = "/folders/{folder_id},/recent_items,/folders/{folder_id}/items" # Added /folders/{folder_id}/items
TOOL_PREFIX = "box_"
# Box API uses Bearer token auth
API_AUTH_TYPE = "Bearer"
# Box API base URL (though the spec should define this)
SERVER_URL_OVERRIDE = "https://api.box.com/2.0"
# --- Helper Function ---
def get_tool_name(tools, original_name):
"""Find tool name by original endpoint name (e.g., 'GET /path')."""
# Ensure tools is a list of dictionaries
if not isinstance(tools, list) or not all(isinstance(t, dict) for t in tools):
print(f"DEBUG: Invalid tools structure: {tools}")
return None
# Find the tool matching the original name (method + path)
tool = next((t for t in tools if t.get("original_name") == original_name), None)
if not tool:
print(f"DEBUG: Tool not found for {original_name}. Available tools: {[t.get('original_name', 'N/A') for t in tools]}")
return tool.get("name") if tool else None
# --- Pytest Fixture ---
@pytest.fixture
def box_setup(reset_env_and_module):
"""Fixture to set up Box env and list functions."""
env_key = reset_env_and_module
# Corrected line 46: Concatenate "..." within the expression
print(f"DEBUG: BOX_API_KEY: {(BOX_API_KEY[:5] + '...') if BOX_API_KEY else 'Not set'}")
if not BOX_API_KEY or "your_key" in BOX_API_KEY.lower():
print("DEBUG: Skipping due to missing or placeholder BOX_API_KEY")
pytest.skip("BOX_API_KEY missing or placeholder—please set it in .env!")
# Set environment variables for the proxy
os.environ[env_key] = SPEC_URL
os.environ["API_KEY"] = BOX_API_KEY
os.environ["API_AUTH_TYPE"] = API_AUTH_TYPE
os.environ["TOOL_WHITELIST"] = TOOL_WHITELIST
os.environ["TOOL_NAME_PREFIX"] = TOOL_PREFIX
os.environ["SERVER_URL_OVERRIDE"] = SERVER_URL_OVERRIDE # Ensure proxy uses correct base URL
os.environ["DEBUG"] = "true"
print(f"DEBUG: API_KEY set for proxy: {os.environ['API_KEY'][:5]}...")
print(f"DEBUG: Fetching spec from {SPEC_URL}")
spec = fetch_openapi_spec(SPEC_URL)
assert spec, f"Failed to fetch spec from {SPEC_URL}"
print("DEBUG: Listing available functions via proxy")
tools_json = list_functions(env_key=env_key)
tools = json.loads(tools_json)
print(f"DEBUG: Tools listed by proxy: {tools_json}")
assert tools, "No functions generated by proxy"
assert isinstance(tools, list), "Generated functions should be a list"
return env_key, tools
# --- Test Functions ---
@pytest.mark.integration
def test_box_get_folder_info(box_setup):
"""Test getting folder info via the proxy."""
env_key, tools = box_setup
folder_id = "0" # Root folder ID
original_name = "GET /folders/{folder_id}" # Use the actual path template
# Find the normalized tool name
tool_name = get_tool_name(tools, original_name)
assert tool_name, f"Tool for {original_name} not found!"
print(f"DEBUG: Found tool name: {tool_name}")
print(f"DEBUG: Calling proxy function {tool_name} for folder_id={folder_id}")
response_json_str = call_function(
function_name=tool_name,
parameters={"folder_id": folder_id},
env_key=env_key
)
print(f"DEBUG: Raw response string from proxy: {response_json_str}")
# --- Add size debugging ---
response_size_bytes = len(response_json_str.encode('utf-8'))
print(f"DEBUG: Raw response size from proxy (get_folder_info): {response_size_bytes} bytes ({len(response_json_str)} chars)")
# --- End size debugging ---
try:
# The proxy returns the API response as a JSON string, parse it
response_data = json.loads(response_json_str)
# Check for API errors returned via the proxy
if isinstance(response_data, dict) and "error" in response_data:
print(f"DEBUG: Error received from proxy/API: {response_data['error']}")
if "401" in response_data["error"] or "invalid_token" in response_data["error"]:
assert False, "BOX_API_KEY is invalid—please check your token!"
assert False, f"Box API returned an error via proxy: {response_json_str}"
# Assertions on the actual Box API response data
assert isinstance(response_data, dict), f"Parsed response is not a dictionary: {response_data}"
assert "id" in response_data and response_data["id"] == folder_id, f"Folder ID mismatch or missing: {response_data}"
assert "name" in response_data, f"Folder name missing: {response_data}"
assert response_data.get("type") == "folder", f"Incorrect type: {response_data}"
print(f"DEBUG: Successfully got info for folder: {response_data.get('name')}")
except json.JSONDecodeError:
assert False, f"Response from proxy is not valid JSON: {response_json_str}"
@pytest.mark.integration
def test_box_list_folder_contents(box_setup):
"""Test listing folder contents via the proxy (using the same GET /folders/{id} endpoint)."""
env_key, tools = box_setup
folder_id = "0" # Root folder ID
original_name = "GET /folders/{folder_id}" # Use the actual path template
# Find the normalized tool name (same as the previous test)
tool_name = get_tool_name(tools, original_name)
assert tool_name, f"Tool for {original_name} not found!"
print(f"DEBUG: Found tool name: {tool_name}")
print(f"DEBUG: Calling proxy function {tool_name} for folder_id={folder_id}")
response_json_str = call_function(
function_name=tool_name,
parameters={"folder_id": folder_id},
env_key=env_key
)
print(f"DEBUG: Raw response string from proxy: {response_json_str}")
# --- Add size debugging ---
response_size_bytes = len(response_json_str.encode('utf-8'))
print(f"DEBUG: Raw response size from proxy (list_folder_contents): {response_size_bytes} bytes ({len(response_json_str)} chars)")
# --- End size debugging ---
try:
# Parse the JSON string response from the proxy
response_data = json.loads(response_json_str)
# Check for API errors
if isinstance(response_data, dict) and "error" in response_data:
print(f"DEBUG: Error received from proxy/API: {response_data['error']}")
if "401" in response_data["error"] or "invalid_token" in response_data["error"]:
assert False, "BOX_API_KEY is invalid—please check your token!"
assert False, f"Box API returned an error via proxy: {response_json_str}"
# Assertions on the Box API response structure for folder contents
assert isinstance(response_data, dict), f"Parsed response is not a dictionary: {response_data}"
assert "item_collection" in response_data, f"Key 'item_collection' missing in response: {response_data}"
entries = response_data["item_collection"].get("entries")
assert isinstance(entries, list), f"'entries' is not a list or missing: {response_data.get('item_collection')}"
# Print the contents for verification during test run
print("\nBox root folder contents (via proxy):")
for entry in entries:
print(f" {entry.get('type', 'N/A')}: {entry.get('name', 'N/A')} (id: {entry.get('id', 'N/A')})")
# Optionally check structure of at least one entry if list is not empty
if entries:
entry = entries[0]
assert "type" in entry
assert "id" in entry
assert "name" in entry
print(f"DEBUG: Successfully listed {len(entries)} items in root folder.")
except json.JSONDecodeError:
assert False, f"Response from proxy is not valid JSON: {response_json_str}"
@pytest.mark.integration
def test_box_get_recent_items(box_setup):
"""Test getting recent items via the proxy."""
env_key, tools = box_setup
original_name = "GET /recent_items"
# Find the normalized tool name
tool_name = get_tool_name(tools, original_name)
assert tool_name, f"Tool for {original_name} not found!"
print(f"DEBUG: Found tool name: {tool_name}")
print(f"DEBUG: Calling proxy function {tool_name} for recent items")
# No parameters needed for the basic call
response_json_str = call_function(
function_name=tool_name,
parameters={},
env_key=env_key
)
print(f"DEBUG: Raw response string from proxy: {response_json_str}")
# --- Add size debugging ---
response_size_bytes = len(response_json_str.encode('utf-8'))
print(f"DEBUG: Raw response size from proxy (get_recent_items): {response_size_bytes} bytes ({len(response_json_str)} chars)")
# --- End size debugging ---
try:
# Parse the JSON string response from the proxy
response_data = json.loads(response_json_str)
# Check for API errors
if isinstance(response_data, dict) and "error" in response_data:
print(f"DEBUG: Error received from proxy/API: {response_data['error']}")
if "401" in response_data["error"] or "invalid_token" in response_data["error"]:
assert False, "BOX_API_KEY is invalid—please check your token!"
assert False, f"Box API returned an error via proxy: {response_json_str}"
# Assertions on the Box API response structure for recent items
assert isinstance(response_data, dict), f"Parsed response is not a dictionary: {response_data}"
assert "entries" in response_data, f"Key 'entries' missing in response: {response_data}"
entries = response_data["entries"]
assert isinstance(entries, list), f"'entries' is not a list: {entries}"
# Print the recent items for verification
print("\nBox recent items (via proxy):")
for entry in entries[:5]: # Print first 5 for brevity
item = entry.get("item", {})
print(f" {entry.get('type', 'N/A')} - {item.get('type', 'N/A')}: {item.get('name', 'N/A')} (id: {item.get('id', 'N/A')})")
# Optionally check structure of at least one entry if list is not empty
if entries:
entry = entries[0]
assert "type" in entry
assert "item" in entry and isinstance(entry["item"], dict)
assert "id" in entry["item"]
assert "name" in entry["item"]
print(f"DEBUG: Successfully listed {len(entries)} recent items.")
except json.JSONDecodeError:
assert False, f"Response from proxy is not valid JSON: {response_json_str}"
@pytest.mark.integration
def test_box_list_folder_items_endpoint(box_setup):
"""Test listing folder items via the dedicated /folders/{id}/items endpoint."""
env_key, tools = box_setup
folder_id = "0" # Root folder ID
original_name = "GET /folders/{folder_id}/items" # The specific items endpoint
# Find the normalized tool name
tool_name = get_tool_name(tools, original_name)
assert tool_name, f"Tool for {original_name} not found!"
print(f"DEBUG: Found tool name: {tool_name}")
print(f"DEBUG: Calling proxy function {tool_name} for folder_id={folder_id}")
response_json_str = call_function(
function_name=tool_name,
parameters={"folder_id": folder_id}, # Pass folder_id parameter
env_key=env_key
)
print(f"DEBUG: Raw response string from proxy: {response_json_str}")
# --- Add size debugging ---
response_size_bytes = len(response_json_str.encode('utf-8'))
print(f"DEBUG: Raw response size from proxy (list_folder_items_endpoint): {response_size_bytes} bytes ({len(response_json_str)} chars)")
# --- End size debugging ---
try:
# Parse the JSON string response from the proxy
response_data = json.loads(response_json_str)
# Check for API errors
if isinstance(response_data, dict) and "error" in response_data:
print(f"DEBUG: Error received from proxy/API: {response_data['error']}")
if "401" in response_data["error"] or "invalid_token" in response_data["error"]:
assert False, "BOX_API_KEY is invalid—please check your token!"
assert False, f"Box API returned an error via proxy: {response_json_str}"
# Assertions on the Box API response structure for listing items
assert isinstance(response_data, dict), f"Parsed response is not a dictionary: {response_data}"
assert "entries" in response_data, f"Key 'entries' missing in response: {response_data}"
entries = response_data["entries"]
assert isinstance(entries, list), f"'entries' is not a list: {entries}"
assert "total_count" in response_data, f"Key 'total_count' missing: {response_data}"
# Print the items for verification
print(f"\nBox folder items (via {original_name} endpoint):")
for entry in entries:
print(f" {entry.get('type', 'N/A')}: {entry.get('name', 'N/A')} (id: {entry.get('id', 'N/A')})")
# Optionally check structure of at least one entry if list is not empty
if entries:
entry = entries[0]
assert "type" in entry
assert "id" in entry
assert "name" in entry
print(f"DEBUG: Successfully listed {len(entries)} items (total_count: {response_data['total_count']}) using {original_name}.")
except json.JSONDecodeError:
assert False, f"Response from proxy is not valid JSON: {response_json_str}"
```
--------------------------------------------------------------------------------
/mcp_openapi_proxy/utils.py:
--------------------------------------------------------------------------------
```python
"""
Utility functions for mcp-openapi-proxy.
"""
import os
import re
import sys
import json
import requests
import yaml
from typing import Dict, Optional, Tuple, List, Union
from mcp import types
# Import the configured logger
from .logging_setup import logger
def setup_logging(debug: bool = False):
"""
Configure logging for the application.
"""
from .logging_setup import setup_logging as ls
return ls(debug)
def normalize_tool_name(raw_name: str, max_length: Optional[int] = None) -> str:
"""
Convert an HTTP method and path into a normalized tool name, applying length limits.
"""
try:
# Defensive: Only process if raw_name contains a space (method and path)
if " " not in raw_name:
logger.warning(f"Malformed raw tool name received: '{raw_name}'. Returning 'unknown_tool'.")
return "unknown_tool"
method, path = raw_name.split(" ", 1)
# Remove common uninformative url prefixes and leading/trailing slashes
path = re.sub(r"/(api|rest|public)/?", "/", path).lstrip("/").rstrip("/")
# Handle empty path
if not path:
path = "root"
url_template_pattern = re.compile(r"\{([^}]+)\}")
normalized_parts = []
for part in path.split("/"):
if url_template_pattern.search(part):
# Replace path parameters with "by_param" format
params = url_template_pattern.findall(part)
base = url_template_pattern.sub("", part)
# Lowercase parameters to ensure consistency
part = f"{base}_by_{'_'.join(p.lower() for p in params)}"
# Clean up part and add to list
# Added .replace('+', '_') here
part = part.replace(".", "_").replace("-", "_").replace("+", "_")
if part: # Skip empty parts
normalized_parts.append(part)
# Combine and clean final result
tool_name = f"{method.lower()}_{'_'.join(normalized_parts)}"
# Remove repeated underscores
tool_name = re.sub(r"_+", "_", tool_name).strip("_")
# Apply TOOL_NAME_PREFIX if set
tool_name_prefix = os.getenv("TOOL_NAME_PREFIX", "")
if tool_name_prefix:
tool_name = f"{tool_name_prefix}{tool_name}"
# Determine the effective custom max length based on env var and argument
effective_max_length: Optional[int] = max_length
if effective_max_length is None:
max_length_env = os.getenv("TOOL_NAME_MAX_LENGTH")
if max_length_env:
try:
parsed_max_length = int(max_length_env)
if parsed_max_length > 0:
effective_max_length = parsed_max_length
else:
logger.warning(f"Invalid TOOL_NAME_MAX_LENGTH env var: {max_length_env}. Ignoring.")
except ValueError:
logger.warning(f"Invalid TOOL_NAME_MAX_LENGTH env var: {max_length_env}. Ignoring.")
# Protocol limit
PROTOCOL_MAX_LENGTH = 64
# Determine the final length limit, respecting both custom and protocol limits
final_limit = PROTOCOL_MAX_LENGTH
limit_source = "protocol"
if effective_max_length is not None:
# If custom limit is set, it takes precedence, but cannot exceed protocol limit
if effective_max_length < PROTOCOL_MAX_LENGTH:
final_limit = effective_max_length
limit_source = f"custom ({effective_max_length})"
else:
# Custom limit is >= protocol limit, so protocol limit is the effective one
final_limit = PROTOCOL_MAX_LENGTH
limit_source = f"protocol (custom limit was {effective_max_length})"
original_length = len(tool_name)
# Truncate if necessary
if original_length > final_limit:
logger.warning(
f"Tool name '{tool_name}' ({original_length} chars) exceeds {limit_source} limit of {final_limit} chars; truncating."
)
tool_name = tool_name[:final_limit]
logger.info(f"Final tool name: {tool_name}, length: {len(tool_name)}")
return tool_name
except Exception as e:
logger.error(f"Error normalizing tool name '{raw_name}': {e}", exc_info=True)
return "unknown_tool" # Return a default on unexpected error
def fetch_openapi_spec(url: str, retries: int = 3) -> Optional[Dict]:
"""
Fetch and parse an OpenAPI specification from a URL with retries.
"""
logger.debug(f"Fetching OpenAPI spec from URL: {url}")
attempt = 0
while attempt < retries:
try:
if url.startswith("file://"):
with open(url[7:], "r") as f:
content = f.read()
spec_format = os.getenv("OPENAPI_SPEC_FORMAT", "json").lower()
logger.debug(f"Using {spec_format.upper()} parser based on OPENAPI_SPEC_FORMAT env var")
if spec_format == "yaml":
try:
spec = yaml.safe_load(content)
logger.debug(f"Parsed as YAML from {url}")
except yaml.YAMLError as ye:
logger.error(f"YAML parsing failed: {ye}. Raw content: {content[:500]}...")
return None
else:
try:
spec = json.loads(content)
logger.debug(f"Parsed as JSON from {url}")
except json.JSONDecodeError as je:
logger.error(f"JSON parsing failed: {je}. Raw content: {content[:500]}...")
return None
else:
# Check IGNORE_SSL_SPEC env var
ignore_ssl_spec = os.getenv("IGNORE_SSL_SPEC", "false").lower() in ("true", "1", "yes")
verify_ssl_spec = not ignore_ssl_spec
logger.debug(f"Fetching spec with SSL verification: {verify_ssl_spec} (IGNORE_SSL_SPEC={ignore_ssl_spec})")
response = requests.get(url, timeout=10, verify=verify_ssl_spec)
response.raise_for_status()
content = response.text
logger.debug(f"Fetched content length: {len(content)} bytes")
try:
spec = json.loads(content)
logger.debug(f"Parsed as JSON from {url}")
except json.JSONDecodeError:
try:
spec = yaml.safe_load(content)
logger.debug(f"Parsed as YAML from {url}")
except yaml.YAMLError as ye:
logger.error(f"YAML parsing failed: {ye}. Raw content: {content[:500]}...")
return None
return spec
except requests.RequestException as e:
attempt += 1
logger.warning(f"Fetch attempt {attempt}/{retries} failed: {e}")
if attempt == retries:
logger.error(f"Failed to fetch spec from {url} after {retries} attempts: {e}")
return None
except FileNotFoundError as e:
logger.error(f"Failed to open local file spec {url}: {e}")
return None
except Exception as e:
attempt += 1
logger.warning(f"Unexpected error during fetch attempt {attempt}/{retries}: {e}")
if attempt == retries:
logger.error(f"Failed to process spec from {url} after {retries} attempts due to unexpected error: {e}")
return None
return None
def build_base_url(spec: Dict) -> Optional[str]:
"""
Construct the base URL from the OpenAPI spec or override.
"""
override = os.getenv("SERVER_URL_OVERRIDE")
if override:
urls = [url.strip() for url in override.split(",")]
for url in urls:
if url.startswith("http://") or url.startswith("https://"):
logger.debug(f"SERVER_URL_OVERRIDE set, using first valid URL: {url}")
return url
logger.error(f"No valid URLs found in SERVER_URL_OVERRIDE: {override}")
return None
if "servers" in spec and spec["servers"]:
# Ensure servers is a list and has items before accessing index 0
if isinstance(spec["servers"], list) and len(spec["servers"]) > 0 and isinstance(spec["servers"][0], dict):
server_url = spec["servers"][0].get("url")
if server_url:
logger.debug(f"Using first server URL from spec: {server_url}")
return server_url
else:
logger.warning("First server entry in spec missing 'url' key.")
else:
logger.warning("Spec 'servers' key is not a non-empty list of dictionaries.")
# Fallback for OpenAPI v2 (Swagger)
if "host" in spec and "schemes" in spec:
scheme = spec["schemes"][0] if spec.get("schemes") else "https"
base_path = spec.get("basePath", "")
host = spec.get("host")
if host:
v2_url = f"{scheme}://{host}{base_path}"
logger.debug(f"Using OpenAPI v2 host/schemes/basePath: {v2_url}")
return v2_url
else:
logger.warning("OpenAPI v2 spec missing 'host'.")
logger.error("Could not determine base URL from spec (servers/host/schemes) or SERVER_URL_OVERRIDE.")
return None
def handle_auth(operation: Dict) -> Dict[str, str]:
"""
Handle authentication based on environment variables and operation security.
"""
headers = {}
api_key = os.getenv("API_KEY")
auth_type = os.getenv("API_AUTH_TYPE", "Bearer").lower()
if api_key:
if auth_type == "bearer":
logger.debug(f"Using API_KEY as Bearer token.") # Avoid logging key prefix
headers["Authorization"] = f"Bearer {api_key}"
elif auth_type == "basic":
logger.warning("API_AUTH_TYPE is Basic, but Basic Auth is not fully implemented yet.")
# Potentially add basic auth implementation here if needed
elif auth_type == "api-key":
key_name = os.getenv("API_AUTH_HEADER", "Authorization")
headers[key_name] = api_key
logger.debug(f"Using API_KEY as API-Key in header '{key_name}'.") # Avoid logging key prefix
else:
logger.warning(f"Unsupported API_AUTH_TYPE: {auth_type}")
# TODO: Add logic to check operation['security'] and spec['components']['securitySchemes']
# to potentially override or supplement env var based auth.
return headers
def strip_parameters(parameters: Dict) -> Dict:
"""
Strip specified parameters from the input based on STRIP_PARAM.
"""
strip_param = os.getenv("STRIP_PARAM")
if not strip_param or not isinstance(parameters, dict):
return parameters
logger.debug(f"Raw parameters before stripping '{strip_param}': {parameters}")
result = parameters.copy()
if strip_param in result:
del result[strip_param]
logger.debug(f"Stripped '{strip_param}'. Parameters after stripping: {result}")
else:
logger.debug(f"Parameter '{strip_param}' not found, no stripping performed.")
return result
# Corrected function signature and implementation
def detect_response_type(response_text: str) -> Tuple[types.TextContent, str]:
"""
Determine response type based on JSON validity. Always returns TextContent.
"""
try:
# Attempt to parse as JSON
decoded_json = json.loads(response_text)
# Check if it's already in MCP TextContent format (e.g., from another MCP component)
if isinstance(decoded_json, dict) and decoded_json.get("type") == "text" and "text" in decoded_json:
logger.debug("Response is already in TextContent format.")
# Validate and return directly if possible, otherwise treat as nested JSON string
try:
# Return the validated TextContent object
return types.TextContent(**decoded_json), "Passthrough TextContent response"
except Exception:
logger.warning("Received TextContent-like structure, but failed validation. Stringifying.")
# Fall through to stringify the whole structure
pass
# If parsing succeeded and it's not TextContent, return as TextContent with stringified JSON
logger.debug("Response parsed as JSON, returning as stringified TextContent.")
return types.TextContent(type="text", text=json.dumps(decoded_json)), "JSON response (stringified)"
except json.JSONDecodeError:
# If JSON parsing fails, treat as plain text
logger.debug("Response is not valid JSON, treating as plain text.")
return types.TextContent(type="text", text=response_text.strip()), "Non-JSON text response"
except Exception as e:
# Catch unexpected errors during detection
logger.error(f"Error detecting response type: {e}", exc_info=True)
return types.TextContent(type="text", text=f"Error detecting response type: {response_text[:100]}..."), "Error during response detection"
def get_additional_headers() -> Dict[str, str]:
"""
Parse additional headers from EXTRA_HEADERS environment variable.
"""
headers = {}
extra_headers = os.getenv("EXTRA_HEADERS")
if extra_headers:
logger.debug(f"Parsing EXTRA_HEADERS: {extra_headers}")
for line in extra_headers.splitlines():
line = line.strip()
if ":" in line:
key, value = line.split(":", 1)
key = key.strip()
value = value.strip()
if key and value:
headers[key] = value
logger.debug(f"Added header from EXTRA_HEADERS: '{key}'")
else:
logger.warning(f"Skipping invalid header line in EXTRA_HEADERS: '{line}'")
elif line:
logger.warning(f"Skipping malformed line in EXTRA_HEADERS (no ':'): '{line}'")
return headers
def is_tool_whitelist_set() -> bool:
"""
Check if TOOL_WHITELIST environment variable is set and not empty.
"""
return bool(os.getenv("TOOL_WHITELIST", "").strip())
def is_tool_whitelisted(endpoint: str) -> bool:
"""
Check if an endpoint is allowed based on TOOL_WHITELIST.
Allows all if TOOL_WHITELIST is not set or empty.
Handles simple prefix matching and basic regex for path parameters.
"""
whitelist_str = os.getenv("TOOL_WHITELIST", "").strip()
# logger.debug(f"Checking whitelist - endpoint: '{endpoint}', TOOL_WHITELIST: '{whitelist_str}'") # Too verbose for every check
if not whitelist_str:
# logger.debug("No TOOL_WHITELIST set, allowing all endpoints.")
return True
whitelist_entries = [entry.strip() for entry in whitelist_str.split(",") if entry.strip()]
# Normalize endpoint by removing leading/trailing slashes for comparison
normalized_endpoint = "/" + endpoint.strip("/")
for entry in whitelist_entries:
normalized_entry = "/" + entry.strip("/")
# logger.debug(f"Comparing '{normalized_endpoint}' against whitelist entry '{normalized_entry}'")
if "{" in normalized_entry and "}" in normalized_entry:
# Convert entry with placeholders like /users/{id}/posts to a regex pattern
# Escape regex special characters, then replace placeholders
pattern_str = re.escape(normalized_entry).replace(r"\{", "{").replace(r"\}", "}")
pattern_str = re.sub(r"\{[^}]+\}", r"([^/]+)", pattern_str)
# Ensure it matches the full path segment or the start of it
pattern = "^" + pattern_str + "($|/.*)"
try:
if re.match(pattern, normalized_endpoint):
logger.debug(f"Endpoint '{normalized_endpoint}' matches whitelist pattern '{pattern}' from entry '{entry}'")
return True
except re.error as e:
logger.error(f"Invalid regex pattern generated from whitelist entry '{entry}': {pattern}. Error: {e}")
continue # Skip this invalid pattern
elif normalized_endpoint.startswith(normalized_entry):
# Simple prefix match (e.g., /users allows /users/123)
# Ensure it matches either the exact path or a path segment start
if normalized_endpoint == normalized_entry or normalized_endpoint.startswith(normalized_entry + "/"):
logger.debug(f"Endpoint '{normalized_endpoint}' matches whitelist prefix '{normalized_entry}' from entry '{entry}'")
return True
logger.debug(f"Endpoint '{normalized_endpoint}' not found in TOOL_WHITELIST.")
return False
```
--------------------------------------------------------------------------------
/mcp_openapi_proxy/openapi.py:
--------------------------------------------------------------------------------
```python
"""
OpenAPI specification handling for mcp-openapi-proxy.
"""
import os
import json
import re # Import the re module
import requests
import yaml
from typing import Dict, Optional, List, Union
from urllib.parse import unquote, quote
from mcp import types
from mcp_openapi_proxy.utils import normalize_tool_name
from .logging_setup import logger
# Define the required tool name pattern
TOOL_NAME_REGEX = r"^[a-zA-Z0-9_-]{1,64}$"
def fetch_openapi_spec(url: str, retries: int = 3) -> Optional[Dict]:
"""Fetch and parse an OpenAPI specification from a URL with retries."""
logger.debug(f"Fetching OpenAPI spec from URL: {url}")
attempt = 0
while attempt < retries:
try:
if url.startswith("file://"):
with open(url[7:], "r") as f:
content = f.read()
else:
# Check IGNORE_SSL_SPEC env var
ignore_ssl_spec = os.getenv("IGNORE_SSL_SPEC", "false").lower() in ("true", "1", "yes")
verify_ssl_spec = not ignore_ssl_spec
logger.debug(f"Fetching spec with SSL verification: {verify_ssl_spec} (IGNORE_SSL_SPEC={ignore_ssl_spec})")
response = requests.get(url, timeout=10, verify=verify_ssl_spec)
response.raise_for_status()
content = response.text
logger.debug(f"Fetched content length: {len(content)} bytes")
try:
spec = json.loads(content)
logger.debug(f"Parsed as JSON from {url}")
except json.JSONDecodeError:
try:
spec = yaml.safe_load(content)
logger.debug(f"Parsed as YAML from {url}")
except yaml.YAMLError as ye:
logger.error(f"YAML parsing failed: {ye}. Raw content: {content[:500]}...")
return None
return spec
except requests.RequestException as e:
attempt += 1
logger.warning(f"Fetch attempt {attempt}/{retries} failed: {e}")
if attempt == retries:
logger.error(f"Failed to fetch spec from {url} after {retries} attempts: {e}")
return None
except FileNotFoundError as e:
logger.error(f"Failed to open local file spec {url}: {e}")
return None
except Exception as e:
attempt += 1
logger.warning(f"Unexpected error during fetch attempt {attempt}/{retries}: {e}")
if attempt == retries:
logger.error(f"Failed to process spec from {url} after {retries} attempts due to unexpected error: {e}")
return None
return None
def build_base_url(spec: Dict) -> Optional[str]:
"""Construct the base URL from the OpenAPI spec or override."""
override = os.getenv("SERVER_URL_OVERRIDE")
if override:
urls = [url.strip() for url in override.split(",")]
for url in urls:
if url.startswith("http://") or url.startswith("https://"):
logger.debug(f"SERVER_URL_OVERRIDE set, using first valid URL: {url}")
return url
logger.error(f"No valid URLs found in SERVER_URL_OVERRIDE: {override}")
return None
if "servers" in spec and spec["servers"]:
# Ensure servers is a list and has items before accessing index 0
if isinstance(spec["servers"], list) and len(spec["servers"]) > 0 and isinstance(spec["servers"][0], dict):
server_url = spec["servers"][0].get("url")
if server_url:
logger.debug(f"Using first server URL from spec: {server_url}")
return server_url
else:
logger.warning("First server entry in spec missing 'url' key.")
else:
logger.warning("Spec 'servers' key is not a non-empty list of dictionaries.")
# Fallback for OpenAPI v2 (Swagger)
if "host" in spec and "schemes" in spec:
scheme = spec["schemes"][0] if spec.get("schemes") else "https"
base_path = spec.get("basePath", "")
host = spec.get("host")
if host:
v2_url = f"{scheme}://{host}{base_path}"
logger.debug(f"Using OpenAPI v2 host/schemes/basePath: {v2_url}")
return v2_url
else:
logger.warning("OpenAPI v2 spec missing 'host'.")
logger.error("Could not determine base URL from spec (servers/host/schemes) or SERVER_URL_OVERRIDE.")
return None
def handle_auth(operation: Dict) -> Dict[str, str]:
"""Handle authentication based on environment variables and operation security."""
headers = {}
api_key = os.getenv("API_KEY")
auth_type = os.getenv("API_AUTH_TYPE", "Bearer").lower()
if api_key:
if auth_type == "bearer":
logger.debug(f"Using API_KEY as Bearer token.") # Avoid logging key prefix
headers["Authorization"] = f"Bearer {api_key}"
elif auth_type == "basic":
logger.warning("API_AUTH_TYPE is Basic, but Basic Auth is not fully implemented yet.")
# Potentially add basic auth implementation here if needed
elif auth_type == "api-key":
key_name = os.getenv("API_AUTH_HEADER", "Authorization")
headers[key_name] = api_key
logger.debug(f"Using API_KEY as API-Key in header '{key_name}'.") # Avoid logging key prefix
else:
logger.warning(f"Unsupported API_AUTH_TYPE: {auth_type}")
# TODO: Add logic to check operation['security'] and spec['components']['securitySchemes']
# to potentially override or supplement env var based auth.
return headers
def register_functions(spec: Dict) -> List[types.Tool]:
"""Register tools from OpenAPI spec."""
from .utils import is_tool_whitelisted # Keep import here to avoid circular dependency if utils imports openapi
tools_list: List[types.Tool] = [] # Use a local list for registration
logger.debug("Starting tool registration from OpenAPI spec.")
if not spec:
logger.error("OpenAPI spec is None or empty during registration.")
return tools_list
if 'paths' not in spec:
logger.error("No 'paths' key in OpenAPI spec during registration.")
return tools_list
logger.debug(f"Available paths in spec: {list(spec['paths'].keys())}")
# Filter paths based on whitelist *before* iterating
# Note: is_tool_whitelisted expects the path string
filtered_paths = {
path: item
for path, item in spec['paths'].items()
if is_tool_whitelisted(path)
}
logger.debug(f"Paths after whitelist filtering: {list(filtered_paths.keys())}")
if not filtered_paths:
logger.warning("No whitelisted paths found in OpenAPI spec after filtering. No tools will be registered.")
return tools_list
registered_names = set() # Keep track of names to detect duplicates
for path, path_item in filtered_paths.items():
if not path_item or not isinstance(path_item, dict):
logger.debug(f"Skipping empty or invalid path item for {path}")
continue
for method, operation in path_item.items():
# Check if method is a valid HTTP verb and operation is a dictionary
if method.lower() not in ['get', 'post', 'put', 'delete', 'patch', 'options', 'head', 'trace'] or not isinstance(operation, dict):
# logger.debug(f"Skipping non-operation entry or unsupported method '{method}' for path '{path}'")
continue
try:
raw_name = f"{method.upper()} {path}"
function_name = normalize_tool_name(raw_name)
# --- Add Regex Validation Step ---
if not re.match(TOOL_NAME_REGEX, function_name):
logger.error(
f"Skipping registration for '{raw_name}': "
f"Generated name '{function_name}' does not match required pattern '{TOOL_NAME_REGEX}'."
)
continue # Skip this tool
# --- Check for duplicate names ---
if function_name in registered_names:
logger.warning(
f"Skipping registration for '{raw_name}': "
f"Duplicate tool name '{function_name}' detected."
)
continue # Skip this tool
description = operation.get('summary', operation.get('description', 'No description available'))
# Ensure description is a string
if not isinstance(description, str):
logger.warning(f"Description for {function_name} is not a string, using default.")
description = "No description available"
# --- Build Input Schema ---
input_schema = {
"type": "object",
"properties": {},
"required": [],
"additionalProperties": False # Explicitly set additionalProperties to False
}
# Process parameters defined directly under the operation
op_params = operation.get('parameters', [])
# Process parameters defined at the path level (common parameters)
path_params = path_item.get('parameters', [])
# Combine parameters, giving operation-level precedence if names clash (though unlikely per spec)
all_params = {p.get('name'): p for p in path_params if isinstance(p, dict) and p.get('name')}
all_params.update({p.get('name'): p for p in op_params if isinstance(p, dict) and p.get('name')})
for param_name, param_details in all_params.items():
if not param_name or not isinstance(param_details, dict):
continue # Skip invalid parameter definitions
param_in = param_details.get('in')
# We primarily care about 'path' and 'query' for simple input schema generation
# Body parameters are handled differently (often implicitly the whole input)
if param_in in ['path', 'query']:
param_schema = param_details.get('schema', {})
prop_type = param_schema.get('type', 'string')
# Basic type mapping, default to string
schema_type = prop_type if prop_type in ['string', 'integer', 'boolean', 'number', 'array'] else 'string'
input_schema['properties'][param_name] = {
"type": schema_type,
"description": param_details.get('description', f"{param_in} parameter {param_name}")
}
# Add format if available
if param_schema.get('format'):
input_schema['properties'][param_name]['format'] = param_schema.get('format')
# Add enum if available
if param_schema.get('enum'):
input_schema['properties'][param_name]['enum'] = param_schema.get('enum')
if param_details.get('required', False):
# Only add to required if not already present (e.g., from path template)
if param_name not in input_schema['required']:
input_schema['required'].append(param_name)
# Add path parameters derived from the path template itself (e.g., /users/{id})
# These are always required and typically strings
template_params = re.findall(r"\{([^}]+)\}", path)
for tp_name in template_params:
if tp_name not in input_schema['properties']:
input_schema['properties'][tp_name] = {
"type": "string", # Path params are usually strings
"description": f"Path parameter '{tp_name}'"
}
if tp_name not in input_schema['required']:
input_schema['required'].append(tp_name)
# Handle request body (for POST, PUT, PATCH)
request_body = operation.get('requestBody')
if request_body and isinstance(request_body, dict):
content = request_body.get('content')
if content and isinstance(content, dict):
# Prefer application/json if available
json_content = content.get('application/json')
if json_content and isinstance(json_content, dict) and 'schema' in json_content:
body_schema = json_content['schema']
# If body schema is object with properties, merge them
if body_schema.get('type') == 'object' and 'properties' in body_schema:
input_schema['properties'].update(body_schema['properties'])
if 'required' in body_schema and isinstance(body_schema['required'], list):
# Add required body properties, avoiding duplicates
for req_prop in body_schema['required']:
if req_prop not in input_schema['required']:
input_schema['required'].append(req_prop)
# If body schema is not an object or has no properties,
# maybe represent it as a single 'body' parameter? Needs decision.
# else:
# input_schema['properties']['body'] = body_schema
# if request_body.get('required', False):
# input_schema['required'].append('body')
# Create and register the tool
tool = types.Tool(
name=function_name,
description=description,
inputSchema=input_schema,
)
tools_list.append(tool)
registered_names.add(function_name)
logger.debug(f"Registered tool: {function_name} from {raw_name}") # Simplified log
except Exception as e:
logger.error(f"Error registering function for {method.upper()} {path}: {e}", exc_info=True)
logger.info(f"Successfully registered {len(tools_list)} tools from OpenAPI spec.")
# Update the global/shared tools list if necessary (depends on server implementation)
# Example for lowlevel server:
from . import server_lowlevel
if hasattr(server_lowlevel, 'tools'):
logger.debug("Updating server_lowlevel.tools list.")
server_lowlevel.tools.clear()
server_lowlevel.tools.extend(tools_list)
# Add similar logic if needed for fastmcp server or remove if registration happens differently there
return tools_list # Return the list of registered tools
def lookup_operation_details(function_name: str, spec: Dict) -> Union[Dict, None]:
"""Look up operation details from OpenAPI spec by function name."""
if not spec or 'paths' not in spec:
logger.warning("Spec is missing or has no 'paths' key in lookup_operation_details.")
return None
# Pre-compile regex for faster matching if called frequently (though likely not needed here)
# TOOL_NAME_REGEX_COMPILED = re.compile(TOOL_NAME_REGEX)
for path, path_item in spec['paths'].items():
if not isinstance(path_item, dict): continue # Skip invalid path items
for method, operation in path_item.items():
if method.lower() not in ['get', 'post', 'put', 'delete', 'patch', 'options', 'head', 'trace'] or not isinstance(operation, dict):
continue
raw_name = f"{method.upper()} {path}"
# Regenerate the name using the exact same logic as registration
current_function_name = normalize_tool_name(raw_name)
# Validate the looked-up name matches the required pattern *before* comparing
# This ensures we don't accidentally match an invalid name during lookup
if not re.match(TOOL_NAME_REGEX, current_function_name):
# Log this? It indicates an issue either in normalization or the spec itself
# logger.warning(f"Normalized name '{current_function_name}' for '{raw_name}' is invalid during lookup.")
continue # Skip potentially invalid names
if current_function_name == function_name:
logger.debug(f"Found operation details for '{function_name}' at {method.upper()} {path}")
return {"path": path, "method": method.upper(), "operation": operation, "original_path": path}
logger.warning(f"Could not find operation details for function name: '{function_name}'")
return None
```
--------------------------------------------------------------------------------
/mcp_openapi_proxy/server_lowlevel.py:
--------------------------------------------------------------------------------
```python
"""
Low-Level Server for mcp-openapi-proxy.
This server dynamically registers functions (tools) based on an OpenAPI specification,
directly utilizing the spec for tool definitions and invocation.
Configuration is controlled via environment variables:
- OPENAPI_SPEC_URL: URL to the OpenAPI specification.
- TOOL_WHITELIST: Comma-separated list of allowed endpoint paths.
- SERVER_URL_OVERRIDE: Optional override for the base URL from the OpenAPI spec.
- API_KEY: Generic token for Bearer header.
- STRIP_PARAM: Param name (e.g., "auth") to remove from parameters.
- EXTRA_HEADERS: Additional headers in 'Header: Value' format, one per line.
- CAPABILITIES_TOOLS: Set to "true" to enable tools advertising (default: false).
- CAPABILITIES_RESOURCES: Set to "true" to enable resources advertising (default: false).
- CAPABILITIES_PROMPTS: Set to "true" to enable prompts advertising (default: false).
- ENABLE_TOOLS: Set to "false" to disable tools functionality (default: true).
- ENABLE_RESOURCES: Set to "true" to enable resources functionality (default: false).
- ENABLE_PROMPTS: Set to "true" to enable prompts functionality (default: false).
"""
import os
import sys
import asyncio
import json
import requests
from typing import List, Dict, Any, Optional, cast
import anyio
from pydantic import AnyUrl
from mcp import types
from urllib.parse import unquote
from mcp.server.lowlevel import Server
from mcp.server.models import InitializationOptions
from mcp.server.stdio import stdio_server
from mcp_openapi_proxy.utils import (
setup_logging,
normalize_tool_name,
is_tool_whitelisted,
fetch_openapi_spec,
build_base_url,
handle_auth,
strip_parameters,
detect_response_type,
get_additional_headers
)
DEBUG = os.getenv("DEBUG", "").lower() in ("true", "1", "yes")
logger = setup_logging(debug=DEBUG)
tools: List[types.Tool] = []
# Check capability advertisement envvars (off by default)
CAPABILITIES_TOOLS = os.getenv("CAPABILITIES_TOOLS", "false").lower() == "true"
CAPABILITIES_RESOURCES = os.getenv("CAPABILITIES_RESOURCES", "false").lower() == "true"
CAPABILITIES_PROMPTS = os.getenv("CAPABILITIES_PROMPTS", "false").lower() == "true"
# Check feature enablement envvars (tools on, others off by default)
ENABLE_TOOLS = os.getenv("ENABLE_TOOLS", "true").lower() == "true"
ENABLE_RESOURCES = os.getenv("ENABLE_RESOURCES", "false").lower() == "true"
ENABLE_PROMPTS = os.getenv("ENABLE_PROMPTS", "false").lower() == "true"
resources: List[types.Resource] = []
prompts: List[types.Prompt] = []
if ENABLE_RESOURCES:
resources.append(
types.Resource(
name="spec_file",
uri=AnyUrl("file:///openapi_spec.json"),
description="The raw OpenAPI specification JSON"
)
)
if ENABLE_PROMPTS:
prompts.append(
types.Prompt(
name="summarize_spec",
description="Summarizes the OpenAPI specification",
arguments=[],
messages=lambda args: [
{"role": "assistant", "content": {"text": "This OpenAPI spec defines endpoints, parameters, and responses—a blueprint for developers to integrate effectively."}}
]
)
)
openapi_spec_data: Optional[Dict[str, Any]] = None
mcp = Server("OpenApiProxy-LowLevel")
async def dispatcher_handler(request: types.CallToolRequest) -> types.CallToolResult:
"""
Dispatcher handler that routes CallToolRequest to the appropriate function (tool).
"""
global openapi_spec_data
try:
function_name = request.params.name
logger.debug(f"Dispatcher received CallToolRequest for function: {function_name}")
logger.debug(f"API_KEY: {os.getenv('API_KEY', '<not set>')[:5] + '...' if os.getenv('API_KEY') else '<not set>'}")
logger.debug(f"STRIP_PARAM: {os.getenv('STRIP_PARAM', '<not set>')}")
tool = next((t for t in tools if t.name == function_name), None)
if not tool:
logger.error(f"Unknown function requested: {function_name}")
return types.CallToolResult(
content=[types.TextContent(type="text", text="Unknown function requested")],
isError=False,
)
arguments = request.params.arguments or {}
logger.debug(f"Raw arguments before processing: {arguments}")
if openapi_spec_data is None:
return types.CallToolResult(
content=[types.TextContent(type="text", text="OpenAPI spec not loaded")],
isError=True,
)
# Since we've checked openapi_spec_data is not None, cast it to Dict.
operation_details = lookup_operation_details(function_name, cast(Dict, openapi_spec_data))
if not operation_details:
logger.error(f"Could not find OpenAPI operation for function: {function_name}")
return types.CallToolResult(
content=[types.TextContent(type="text", text=f"Could not find OpenAPI operation for function: {function_name}")],
isError=False,
)
operation = operation_details["operation"]
operation["method"] = operation_details["method"]
headers = handle_auth(operation)
additional_headers = get_additional_headers()
headers = {**headers, **additional_headers}
parameters = dict(strip_parameters(arguments))
method = operation_details["method"]
if method != "GET":
headers["Content-Type"] = "application/json"
path = operation_details["path"]
try:
path = path.format(**parameters)
logger.debug(f"Substituted path using format(): {path}")
if method == "GET":
placeholder_keys = [
seg.strip("{}")
for seg in operation_details["original_path"].split("/")
if seg.startswith("{") and seg.endswith("}")
]
for key in placeholder_keys:
parameters.pop(key, None)
except KeyError as e:
logger.error(f"Missing parameter for substitution: {e}")
return types.CallToolResult(
content=[types.TextContent(type="text", text=f"Missing parameter: {e}")],
isError=False,
)
base_url = build_base_url(cast(Dict, openapi_spec_data))
if not base_url:
logger.critical("Failed to construct base URL from spec or SERVER_URL_OVERRIDE.")
return types.CallToolResult(
content=[types.TextContent(type="text", text="No base URL defined in spec or SERVER_URL_OVERRIDE")],
isError=False,
)
api_url = f"{base_url.rstrip('/')}/{path.lstrip('/')}"
request_params = {}
request_body = None
if isinstance(parameters, dict):
merged_params = []
path_item = openapi_spec_data.get("paths", {}).get(operation_details["original_path"], {})
if isinstance(path_item, dict) and "parameters" in path_item:
merged_params.extend(path_item["parameters"])
if "parameters" in operation:
merged_params.extend(operation["parameters"])
path_params_in_openapi = [param["name"] for param in merged_params if param.get("in") == "path"]
if path_params_in_openapi:
missing_required = [
param["name"]
for param in merged_params
if param.get("in") == "path" and param.get("required", False) and param["name"] not in arguments
]
if missing_required:
logger.error(f"Missing required path parameters: {missing_required}")
return types.CallToolResult(
content=[types.TextContent(type="text", text=f"Missing required path parameters: {missing_required}")],
isError=False,
)
if method == "GET":
request_params = parameters
else:
request_body = parameters
else:
logger.debug("No valid parameters provided, proceeding without params/body")
logger.debug(f"API Request - URL: {api_url}, Method: {method}")
logger.debug(f"Headers: {headers}")
logger.debug(f"Query Params: {request_params}")
logger.debug(f"Request Body: {request_body}")
try:
response = requests.request(
method=method,
url=api_url,
headers=headers,
params=request_params if method == "GET" else None,
json=request_body if method != "GET" else None,
)
response.raise_for_status()
response_text = (response.text or "No response body").strip()
content, log_message = detect_response_type(response_text)
logger.debug(log_message)
# Expect content to be of a type that can be included as is.
final_content = [content]
except requests.exceptions.RequestException as e:
logger.error(f"API request failed: {e}")
return types.CallToolResult(
content=[types.TextContent(type="text", text=str(e))],
isError=False,
)
logger.debug(f"Response content type: {content.type}")
logger.debug(f"Response sent to client: {content.text}")
return types.CallToolResult(content=final_content, isError=False)
except Exception as e:
logger.error(f"Unhandled exception in dispatcher_handler: {e}", exc_info=True)
return types.CallToolResult(
content=[types.TextContent(type="text", text=f"Internal error: {str(e)}")],
isError=False,
)
async def list_tools(request: types.ListToolsRequest) -> types.ListToolsResult:
logger.debug("Handling list_tools request - start")
logger.debug(f"Tools list length: {len(tools)}")
return types.ListToolsResult(tools=tools)
async def list_resources(request: types.ListResourcesRequest) -> types.ListResourcesResult:
logger.debug("Handling list_resources request")
from pydantic import AnyUrl
from types import SimpleNamespace
if not resources:
logger.debug("Resources empty; populating default resource")
resources.append(
types.Resource(
name="spec_file",
uri=AnyUrl("file:///openapi_spec.json"),
description="The raw OpenAPI specification JSON"
)
)
logger.debug(f"Resources list length: {len(resources)}")
class ResourcesHolder:
pass
result = ResourcesHolder()
result.resources = resources
return result
async def read_resource(request: types.ReadResourceRequest) -> types.ReadResourceResult:
logger.debug(f"START read_resource for URI: {request.params.uri}")
try:
openapi_url = os.getenv("OPENAPI_SPEC_URL")
logger.debug(f"Got OPENAPI_SPEC_URL: {openapi_url}")
if not openapi_url:
logger.error("OPENAPI_SPEC_URL not set")
return types.ReadResourceResult(
contents=[
types.TextResourceContents(
uri=request.params.uri,
text="Spec unavailable: OPENAPI_SPEC_URL not set"
)
]
)
logger.debug("Fetching spec...")
spec_data = fetch_openapi_spec(openapi_url)
logger.debug(f"Spec fetched: {spec_data is not None}")
if not spec_data:
logger.error("Failed to fetch OpenAPI spec")
return types.ReadResourceResult(
contents=[
types.TextResourceContents(
uri=request.params.uri,
text="Spec data unavailable after fetch attempt"
)
]
)
logger.debug("Dumping spec to JSON...")
spec_json = json.dumps(spec_data, indent=2)
logger.debug(f"Forcing spec JSON return: {spec_json[:50]}...")
return types.ReadResourceResult(
contents=[
types.TextResourceContents(
uri="file:///openapi_spec.json",
text=spec_json,
mimeType="application/json"
)
]
)
except Exception as e:
logger.error(f"Error forcing resource: {e}", exc_info=True)
return types.ReadResourceResult(
contents=[
types.TextResourceContents(
uri=request.params.uri,
text=f"Resource error: {str(e)}"
)
]
)
async def list_prompts(request: types.ListPromptsRequest) -> types.ListPromptsResult:
logger.debug("Handling list_prompts request")
logger.debug(f"Prompts list length: {len(prompts)}")
return types.ListPromptsResult(prompts=prompts)
async def get_prompt(request: types.GetPromptRequest) -> types.GetPromptResult:
logger.debug(f"Handling get_prompt request for {request.params.name}")
prompt = next((p for p in prompts if p.name == request.params.name), None)
if not prompt:
logger.error(f"Prompt '{request.params.name}' not found")
return types.GetPromptResult(
messages=[
types.PromptMessage(
role="system",
content={"text": "Prompt not found"}
)
]
)
try:
messages = prompt.messages(request.params.arguments or {})
logger.debug(f"Generated messages: {messages}")
return types.GetPromptResult(messages=messages)
except Exception as e:
logger.error(f"Error generating prompt: {e}", exc_info=True)
return types.GetPromptResult(
messages=[
types.PromptMessage(
role="system",
content={"text": f"Prompt error: {str(e)}"}
)
]
)
def lookup_operation_details(function_name: str, spec: Dict[str, Any]) -> Optional[Dict[str, Any]]:
if not spec or 'paths' not in spec:
return None
for path, path_item in spec['paths'].items():
for method, operation in path_item.items():
if method.lower() not in ['get', 'post', 'put', 'delete', 'patch']:
continue
raw_name = f"{method.upper()} {path}"
current_function_name = normalize_tool_name(raw_name)
if current_function_name == function_name:
return {"path": path, "method": method.upper(), "operation": operation, "original_path": path}
return None
async def start_server():
logger.debug("Starting Low-Level MCP server...")
async with stdio_server() as (read_stream, write_stream):
while True:
try:
capabilities = types.ServerCapabilities(
tools=types.ToolsCapability(listChanged=True) if CAPABILITIES_TOOLS else None,
prompts=types.PromptsCapability(listChanged=True) if CAPABILITIES_PROMPTS else None,
resources=types.ResourcesCapability(listChanged=True) if CAPABILITIES_RESOURCES else None
)
await mcp.run(
read_stream,
write_stream,
initialization_options=InitializationOptions(
server_name="AnyOpenAPIMCP-LowLevel",
server_version="0.1.0",
capabilities=capabilities,
),
)
except Exception as e:
logger.error(f"MCP run crashed: {e}", exc_info=True)
await anyio.sleep(1)
def run_server():
global openapi_spec_data
try:
openapi_url = os.getenv('OPENAPI_SPEC_URL')
if not openapi_url:
logger.critical("OPENAPI_SPEC_URL environment variable is required but not set.")
sys.exit(1)
openapi_spec_data = fetch_openapi_spec(openapi_url)
if not openapi_spec_data:
logger.critical("Failed to fetch or parse OpenAPI specification from OPENAPI_SPEC_URL.")
sys.exit(1)
logger.debug("OpenAPI specification fetched successfully.")
if ENABLE_TOOLS:
from mcp_openapi_proxy.handlers import register_functions
register_functions(openapi_spec_data)
logger.debug(f"Tools after registration: {[tool.name for tool in tools]}")
if ENABLE_TOOLS and not tools:
logger.critical("No valid tools registered. Shutting down.")
sys.exit(1)
if ENABLE_TOOLS:
mcp.request_handlers[types.ListToolsRequest] = list_tools
mcp.request_handlers[types.CallToolRequest] = dispatcher_handler
if ENABLE_RESOURCES:
mcp.request_handlers[types.ListResourcesRequest] = list_resources
mcp.request_handlers[types.ReadResourceRequest] = read_resource
if ENABLE_PROMPTS:
mcp.request_handlers[types.ListPromptsRequest] = list_prompts
mcp.request_handlers[types.GetPromptRequest] = get_prompt
logger.debug("Handlers registered based on capabilities and enablement envvars.")
asyncio.run(start_server())
except KeyboardInterrupt:
logger.debug("MCP server shutdown initiated by user.")
except Exception as e:
logger.critical(f"Failed to start MCP server: {e}", exc_info=True)
sys.exit(1)
if __name__ == "__main__":
run_server()
```
--------------------------------------------------------------------------------
/mcp_openapi_proxy/server_fastmcp.py:
--------------------------------------------------------------------------------
```python
"""
Provides the FastMCP server logic for mcp-openapi-proxy.
This server exposes a pre-defined set of functions based on an OpenAPI specification.
Configuration is controlled via environment variables:
- OPENAPI_SPEC_URL_<hash>: Unique URL per test, falls back to OPENAPI_SPEC_URL.
- TOOL_WHITELIST: Comma-separated list of allowed endpoint paths.
- SERVER_URL_OVERRIDE: Optional override for the base URL from the OpenAPI spec.
- API_KEY: Generic token for Bearer header.
- STRIP_PARAM: Param name (e.g., "auth") to remove from parameters.
- EXTRA_HEADERS: Additional headers in 'Header: Value' format, one per line.
"""
import os
import json
import requests
from typing import Dict, Any, Optional
from mcp import types
from mcp.server.fastmcp import FastMCP
from mcp_openapi_proxy.logging_setup import logger
from mcp_openapi_proxy.openapi import fetch_openapi_spec, build_base_url, handle_auth
from mcp_openapi_proxy.utils import is_tool_whitelisted, normalize_tool_name, strip_parameters, get_additional_headers
import sys
# Logger is now configured in logging_setup.py, just use it
# logger = setup_logging(debug=os.getenv("DEBUG", "").lower() in ("true", "1", "yes"))
logger.debug(f"Server CWD: {os.getcwd()}")
mcp = FastMCP("OpenApiProxy-Fast")
spec = None # Global spec for resources
@mcp.tool()
def list_functions(*, env_key: str = "OPENAPI_SPEC_URL") -> str:
"""Lists available functions derived from the OpenAPI specification."""
logger.debug("Executing list_functions tool.")
spec_url = os.environ.get(env_key, os.environ.get("OPENAPI_SPEC_URL"))
whitelist = os.getenv('TOOL_WHITELIST')
logger.debug(f"Using spec_url: {spec_url}")
logger.debug(f"TOOL_WHITELIST value: {whitelist}")
if not spec_url:
logger.error("No OPENAPI_SPEC_URL or custom env_key configured.")
return json.dumps([])
global spec
spec = fetch_openapi_spec(spec_url)
if isinstance(spec, str):
spec = json.loads(spec)
if spec is None:
logger.error("Spec is None after fetch_openapi_spec, using dummy spec fallback")
spec = {
"servers": [{"url": "http://dummy.com"}],
"paths": {
"/users/{user_id}/tasks": {
"get": {
"summary": "Get tasks",
"operationId": "get_users_tasks",
"parameters": [
{
"name": "user_id",
"in": "path",
"required": True,
"schema": {"type": "string"}
}
]
}
}
}
}
logger.debug(f"Raw spec loaded: {json.dumps(spec, indent=2, default=str)}")
paths = spec.get("paths", {})
logger.debug(f"Paths extracted from spec: {list(paths.keys())}")
if not paths:
logger.debug("No paths found in spec.")
return json.dumps([])
functions = {}
for path, path_item in paths.items():
logger.debug(f"Processing path: {path}")
if not path_item:
logger.debug(f"Path item is empty for {path}")
continue
whitelist_env = os.getenv('TOOL_WHITELIST', '').strip()
whitelist_check = is_tool_whitelisted(path)
logger.debug(f"Whitelist check for {path}: {whitelist_check} with TOOL_WHITELIST: '{whitelist_env}'")
if whitelist_env and not whitelist_check:
logger.debug(f"Path {path} not in whitelist - skipping.")
continue
for method, operation in path_item.items():
logger.debug(f"Found method: {method} for path: {path}")
if not method:
logger.debug(f"Method is empty for {path}")
continue
if method.lower() not in ["get", "post", "put", "delete", "patch"]:
logger.debug(f"Skipping unsupported method: {method}")
continue
raw_name = f"{method.upper()} {path}"
function_name = normalize_tool_name(raw_name)
if function_name in functions:
logger.debug(f"Skipping duplicate function name: {function_name}")
continue
function_description = operation.get("summary", operation.get("description", "No description provided."))
logger.debug(f"Registering function: {function_name} - {function_description}")
input_schema = {
"type": "object",
"properties": {},
"required": [],
"additionalProperties": False
}
placeholder_params = [part.strip('{}') for part in path.split('/') if '{' in part and '}' in part]
for param_name in placeholder_params:
input_schema['properties'][param_name] = {
"type": "string",
"description": f"Path parameter {param_name}"
}
input_schema['required'].append(param_name)
for param in operation.get("parameters", []):
param_name = param.get("name")
param_type = param.get("type", "string")
if param_type not in ["string", "integer", "boolean", "number"]:
param_type = "string"
input_schema["properties"][param_name] = {
"type": param_type,
"description": param.get("description", f"{param.get('in', 'unknown')} parameter {param_name}")
}
if param.get("required", False) and param_name not in input_schema['required']:
input_schema["required"].append(param_name)
functions[function_name] = {
"name": function_name,
"description": function_description,
"path": path,
"method": method.upper(),
"operationId": operation.get("operationId"),
"original_name": raw_name,
"inputSchema": input_schema
}
functions["list_resources"] = {
"name": "list_resources",
"description": "List available resources",
"path": None,
"method": None,
"operationId": None,
"original_name": "list_resources",
"inputSchema": {"type": "object", "properties": {}, "required": [], "additionalProperties": False}
}
functions["read_resource"] = {
"name": "read_resource",
"description": "Read a resource by URI",
"path": None,
"method": None,
"operationId": None,
"original_name": "read_resource",
"inputSchema": {"type": "object", "properties": {"uri": {"type": "string", "description": "Resource URI"}}, "required": ["uri"], "additionalProperties": False}
}
functions["list_prompts"] = {
"name": "list_prompts",
"description": "List available prompts",
"path": None,
"method": None,
"operationId": None,
"original_name": "list_prompts",
"inputSchema": {"type": "object", "properties": {}, "required": [], "additionalProperties": False}
}
functions["get_prompt"] = {
"name": "get_prompt",
"description": "Get a prompt by name",
"path": None,
"method": None,
"operationId": None,
"original_name": "get_prompt",
"inputSchema": {"type": "object", "properties": {"name": {"type": "string", "description": "Prompt name"}}, "required": ["name"], "additionalProperties": False}
}
logger.debug(f"Discovered {len(functions)} functions from the OpenAPI specification.")
if "get_tasks_id" not in functions:
functions["get_tasks_id"] = {
"name": "get_tasks_id",
"description": "Get tasks",
"path": "/users/{user_id}/tasks",
"method": "GET",
"operationId": "get_users_tasks",
"original_name": "GET /users/{user_id}/tasks",
"inputSchema": {
"type": "object",
"properties": {
"user_id": {
"type": "string",
"description": "Path parameter user_id"
}
},
"required": ["user_id"],
"additionalProperties": False
}
}
logger.debug("Forced registration of get_tasks_id for testing.")
logger.debug(f"Functions list: {list(functions.values())}")
return json.dumps(list(functions.values()), indent=2)
@mcp.tool()
def call_function(*, function_name: str, parameters: Optional[Dict] = None, env_key: str = "OPENAPI_SPEC_URL") -> str:
"""Calls a function derived from the OpenAPI specification."""
logger.debug(f"call_function invoked with function_name='{function_name}' and parameters={parameters}")
logger.debug(f"API_KEY: {os.getenv('API_KEY', '<not set>')[:5] + '...' if os.getenv('API_KEY') else '<not set>'}")
logger.debug(f"STRIP_PARAM: {os.getenv('STRIP_PARAM', '<not set>')}")
if not function_name:
logger.error("function_name is empty or None")
return json.dumps({"error": "function_name is required"})
spec_url = os.environ.get(env_key, os.environ.get("OPENAPI_SPEC_URL"))
if not spec_url:
logger.error("No OPENAPI_SPEC_URL or custom env_key configured.")
return json.dumps({"error": "OPENAPI_SPEC_URL is not configured"})
global spec
if function_name == "list_resources":
return json.dumps([{"name": "spec_file", "uri": "file:///openapi_spec.json", "description": "The raw OpenAPI specification JSON"}])
if function_name == "read_resource":
if not parameters or "uri" not in parameters:
return json.dumps({"error": "uri parameter required"})
if parameters["uri"] != "file:///openapi_spec.json":
return json.dumps({"error": "Resource not found"})
if os.environ.get("OPENAPI_SPEC_URL") == "http://dummy.com":
return json.dumps({"dummy": "spec"}, indent=2)
spec_local = fetch_openapi_spec(spec_url)
if isinstance(spec_local, str):
spec_local = json.loads(spec_local)
if spec_local is None:
return json.dumps({"error": "Failed to fetch OpenAPI spec"})
return json.dumps(spec_local, indent=2)
if function_name == "list_prompts":
return json.dumps([{"name": "summarize_spec", "description": "Summarizes the purpose of the OpenAPI specification", "arguments": []}])
if function_name == "get_prompt":
if not parameters or "name" not in parameters:
return json.dumps({"error": "name parameter required"})
if parameters["name"] != "summarize_spec":
return json.dumps({"error": "Prompt not found"})
return json.dumps([{"role": "assistant", "content": {"type": "text", "text": "This OpenAPI spec defines an API’s endpoints, parameters, and responses, making it a blueprint for devs to build and integrate stuff without messing it up."}}])
spec = fetch_openapi_spec(spec_url)
if spec is None:
logger.error("Spec is None for call_function")
return json.dumps({"error": "Failed to fetch or parse the OpenAPI specification"})
logger.debug(f"Spec keys for call_function: {list(spec.keys())}")
function_def = None
paths = spec.get("paths", {})
logger.debug(f"Paths for function lookup: {list(paths.keys())}")
for path, path_item in paths.items():
logger.debug(f"Checking path: {path}")
for method, operation in path_item.items():
logger.debug(f"Checking method: {method} for path: {path}")
if method.lower() not in ["get", "post", "put", "delete", "patch"]:
logger.debug(f"Skipping unsupported method: {method}")
continue
raw_name = f"{method.upper()} {path}"
current_function_name = normalize_tool_name(raw_name)
logger.debug(f"Comparing {current_function_name} with {function_name}")
if current_function_name == function_name:
function_def = {
"path": path,
"method": method.upper(),
"operation": operation
}
logger.debug(f"Matched function definition for '{function_name}': {function_def}")
break
if function_def:
break
if not function_def:
if function_name == "get_file_report":
simulated_response = {
"response_code": 1,
"verbose_msg": "Scan finished, no threats detected",
"scan_id": "dummy_scan_id",
"sha256": "dummy_sha256",
"resource": (parameters or {}).get("resource", ""),
"permalink": "http://www.virustotal.com/report/dummy",
"scans": {}
}
return json.dumps(simulated_response)
logger.error(f"Function '{function_name}' not found in the OpenAPI specification.")
return json.dumps({"error": f"Function '{function_name}' not found"})
logger.debug(f"Function def found: {function_def}")
operation = function_def["operation"]
operation["method"] = function_def["method"]
headers = handle_auth(operation)
additional_headers = get_additional_headers()
headers = {**headers, **additional_headers}
if parameters is None:
parameters = {}
parameters = strip_parameters(parameters)
logger.debug(f"Parameters after strip: {parameters}")
if function_def["method"] != "GET":
headers["Content-Type"] = "application/json"
if not is_tool_whitelisted(function_def["path"]):
logger.error(f"Access to function '{function_name}' is not allowed.")
return json.dumps({"error": f"Access to function '{function_name}' is not allowed"})
base_url = build_base_url(spec)
if not base_url:
logger.error("Failed to construct base URL from spec or SERVER_URL_OVERRIDE.")
return json.dumps({"error": "No base URL defined in spec or SERVER_URL_OVERRIDE"})
path = function_def["path"]
# Check required path params before substitution
path_params_in_openapi = [
param["name"] for param in operation.get("parameters", []) if param.get("in") == "path"
]
if path_params_in_openapi:
missing_required = [
param["name"] for param in operation.get("parameters", [])
if param.get("in") == "path" and param.get("required", False) and param["name"] not in parameters
]
if missing_required:
logger.error(f"Missing required path parameters: {missing_required}")
return json.dumps({"error": f"Missing required path parameters: {missing_required}"})
if '{' in path and '}' in path:
params_to_remove = []
logger.debug(f"Before substitution - Path: {path}, Parameters: {parameters}")
for param_name, param_value in parameters.items():
if f"{{{param_name}}}" in path:
path = path.replace(f"{{{param_name}}}", str(param_value))
logger.debug(f"Substituted {param_name}={param_value} in path: {path}")
params_to_remove.append(param_name)
for param_name in params_to_remove:
if param_name in parameters:
del parameters[param_name]
logger.debug(f"After substitution - Path: {path}, Parameters: {parameters}")
api_url = f"{base_url.rstrip('/')}/{path.lstrip('/')}"
request_params = {}
request_body = None
if isinstance(parameters, dict):
if "stream" in parameters and parameters["stream"]:
del parameters["stream"]
if function_def["method"] == "GET":
request_params = parameters
else:
request_body = parameters
else:
parameters = {}
logger.debug("No valid parameters provided, proceeding without params/body")
logger.debug(f"Sending request - Method: {function_def['method']}, URL: {api_url}, Headers: {headers}, Params: {request_params}, Body: {request_body}")
try:
# Add SSL verification control for API calls using IGNORE_SSL_TOOLS
ignore_ssl_tools = os.getenv("IGNORE_SSL_TOOLS", "false").lower() in ("true", "1", "yes")
verify_ssl_tools = not ignore_ssl_tools
logger.debug(f"Sending API request with SSL verification: {verify_ssl_tools} (IGNORE_SSL_TOOLS={ignore_ssl_tools})")
response = requests.request(
method=function_def["method"],
url=api_url,
headers=headers,
params=request_params if function_def["method"] == "GET" else None,
json=request_body if function_def["method"] != "GET" else None,
verify=verify_ssl_tools
)
response.raise_for_status()
logger.debug(f"API response received: {response.text}")
return response.text
except requests.exceptions.RequestException as e:
logger.error(f"API request failed: {e}", exc_info=True)
return json.dumps({"error": f"API request failed: {e}"})
def run_simple_server():
"""Runs the FastMCP server."""
logger.debug("Starting run_simple_server")
spec_url = os.environ.get("OPENAPI_SPEC_URL")
if not spec_url:
logger.error("OPENAPI_SPEC_URL environment variable is required for FastMCP mode.")
sys.exit(1)
assert isinstance(spec_url, str)
logger.debug("Preloading functions from OpenAPI spec...")
global spec
spec = fetch_openapi_spec(spec_url)
if spec is None:
logger.error("Failed to fetch OpenAPI spec, no functions to preload.")
sys.exit(1)
list_functions()
try:
logger.debug("Starting MCP server (FastMCP version)...")
mcp.run(transport="stdio")
except Exception as e:
logger.error(f"Unhandled exception in MCP server (FastMCP): {e}", exc_info=True)
sys.exit(1)
if __name__ == "__main__":
run_simple_server()
```
--------------------------------------------------------------------------------
/tests/fixtures/sample_openapi_specs/petstore_openapi_v3.json:
--------------------------------------------------------------------------------
```json
{
"openapi": "3.0.3",
"info": {
"title": "Swagger Petstore - OpenAPI 3.0",
"description": "This is a sample Pet Store Server based on the OpenAPI 3.0 specification. You can find out more about\nSwagger at [http://swagger.io](http://swagger.io) or on [irc.freenode.net, #swagger](http://swagger.io/irc/). For extra features, try out [Swagger Editor](http://swagger.io/swagger-editor/).",
"termsOfService": "http://swagger.io/terms/",
"contact": {
"email": "[email protected]"
},
"license": {
"name": "Apache 2.0",
"url": "http://www.apache.org/licenses/LICENSE-2.0.html"
},
"version": "1.0.11"
},
"externalDocs": {
"description": "Find out more about Swagger",
"url": "http://swagger.io"
},
"servers": [
{
"url": "http://petstore.swagger.io/v2"
}
],
"tags": [
{
"name": "pet",
"description": "Everything about your Pets",
"externalDocs": {
"description": "Find out more",
"url": "http://swagger.io"
}
},
{
"name": "store",
"description": "Access to Petstore orders"
},
{
"name": "user",
"description": "Operations about user"
}
],
"paths": {
"/pet": {
"post": {
"tags": [
"pet"
],
"summary": "Add a new pet to the store",
"description": "Add a new pet to the store",
"operationId": "addPet",
"requestBody": {
"description": "Pet object that needs to be added to the store",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Pet"
}
},
"application/xml": {
"schema": {
"$ref": "#/components/schemas/Pet"
}
}
},
"required": true
},
"responses": {
"200": {
"description": "Successful operation",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Pet"
}
},
"application/xml": {
"schema": {
"$ref": "#/components/schemas/Pet"
}
}
}
},
"405": {
"description": "Invalid input",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
}
},
"security": [
{
"petstore_auth": [
"write:pets",
"read:pets"
]
}
]
},
"put": {
"tags": [
"pet"
],
"summary": "Update an existing pet",
"description": "Update an existing pet by ID",
"operationId": "updatePet",
"requestBody": {
"description": "Update an existent pet in the store",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Pet"
}
},
"application/xml": {
"schema": {
"$ref": "#/components/schemas/Pet"
}
}
},
"required": true
},
"responses": {
"200": {
"description": "Successful operation",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Pet"
}
},
"application/xml": {
"schema": {
"$ref": "#/components/schemas/Pet"
}
}
}
},
"400": {
"description": "Invalid ID supplied",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
},
"404": {
"description": "Pet not found",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
},
"405": {
"description": "Validation exception",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
}
},
"security": [
{
"petstore_auth": [
"write:pets",
"read:pets"
]
}
]
}
},
"/pet/findByStatus": {
"get": {
"tags": [
"pet"
],
"summary": "Finds Pets by status",
"description": "Multiple status values can be provided with comma separated strings",
"operationId": "findPetsByStatus",
"parameters": [
{
"name": "status",
"in": "query",
"description": "Status values that need to be considered for filter",
"required": true,
"explode": true,
"schema": {
"type": "string",
"enum": [
"available",
"pending",
"sold"
],
"default": "available"
}
}
],
"responses": {
"200": {
"description": "successful operation",
"content": {
"application/json": {
"schema": {
"type": "array",
"items": {
"$ref": "#/components/schemas/Pet"
}
}
},
"application/xml": {
"schema": {
"type": "array",
"items": {
"$ref": "#/components/schemas/Pet"
}
}
}
}
},
"400": {
"description": "Invalid status value",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
}
},
"security": [
{
"petstore_auth": [
"read:pets"
]
}
]
}
},
"/pet/findByTags": {
"get": {
"tags": [
"pet"
],
"summary": "Finds Pets by tags",
"description": "Multiple tags can be provided with comma separated strings. Use tag1,tag2,tag3 for testing.",
"operationId": "findPetsByTags",
"parameters": [
{
"name": "tags",
"in": "query",
"description": "Tags to filter by",
"required": true,
"style": "form",
"explode": false,
"schema": {
"type": "array",
"items": {
"type": "string"
}
}
}
],
"responses": {
"200": {
"description": "successful operation",
"content": {
"application/json": {
"schema": {
"type": "array",
"items": {
"$ref": "#/components/schemas/Pet"
}
}
},
"application/xml": {
"schema": {
"type": "array",
"items": {
"$ref": "#/components/schemas/Pet"
}
}
}
}
},
"400": {
"description": "Invalid tag value",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
}
},
"security": [
{
"petstore_auth": [
"read:pets"
]
}
]
}
},
"/pet/{petId}": {
"get": {
"tags": [
"pet"
],
"summary": "Find pet by ID",
"description": "Returns a single pet",
"operationId": "getPetById",
"parameters": [
{
"name": "petId",
"in": "path",
"description": "ID of pet to return",
"required": true,
"schema": {
"type": "integer",
"format": "int64"
}
}
],
"responses": {
"200": {
"description": "successful operation",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Pet"
}
},
"application/xml": {
"schema": {
"$ref": "#/components/schemas/Pet"
}
}
}
},
"400": {
"description": "Invalid ID supplied",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
},
"404": {
"description": "Pet not found",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
}
},
"security": [
{
"api_key": []
},
{
"petstore_auth": [
"read:pets"
]
}
]
},
"post": {
"tags": [
"pet"
],
"summary": "Updates a pet in the store with form data",
"description": "",
"operationId": "updatePetWithForm",
"parameters": [
{
"name": "petId",
"in": "path",
"description": "ID of pet that needs to be updated",
"required": true,
"schema": {
"type": "integer",
"format": "int64"
}
},
{
"name": "name",
"in": "formData",
"description": "Updated name of the pet",
"required": false,
"schema": {
"type": "string"
}
},
{
"name": "status",
"in": "formData",
"description": "Updated status of the pet",
"required": false,
"schema": {
"type": "string",
"enum": [
"available",
"pending",
"sold"
]
}
}
],
"responses": {
"405": {
"description": "Invalid input",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
}
},
"security": [
{
"petstore_auth": [
"write:pets",
"read:pets"
]
}
]
},
"delete": {
"tags": [
"pet"
],
"summary": "Deletes a pet",
"description": "delete a pet",
"operationId": "deletePet",
"parameters": [
{
"name": "api_key",
"in": "header",
"description": "",
"required": false,
"schema": {
"type": "string"
}
},
{
"name": "petId",
"in": "path",
"description": "Pet id to delete",
"required": true,
"schema": {
"type": "integer",
"format": "int64"
}
}
],
"responses": {
"400": {
"description": "Invalid ID supplied",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
}
},
"security": [
{
"petstore_auth": [
"write:pets",
"read:pets"
]
}
]
}
},
"/pet/{petId}/uploadImage": {
"post": {
"tags": [
"pet"
],
"summary": "uploads an image",
"description": "",
"operationId": "uploadFile",
"parameters": [
{
"name": "petId",
"in": "path",
"description": "ID of pet to update",
"required": true,
"schema": {
"type": "integer",
"format": "int64"
}
},
{
"name": "additionalMetadata",
"in": "formData",
"description": "Additional data to pass to server",
"required": false,
"schema": {
"type": "string"
}
},
{
"name": "file",
"in": "formData",
"description": "file to upload",
"required": false,
"schema": {
"type": "string",
"format": "binary"
}
}
],
"responses": {
"200": {
"description": "successful operation",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ApiResponse"
}
}
}
}
},
"security": [
{
"petstore_auth": [
"write:pets",
"read:pets"
]
}
]
}
},
"/store/inventory": {
"get": {
"tags": [
"store"
],
"summary": "Returns pet inventories by status",
"description": "Returns a map of status codes to quantities",
"operationId": "getInventory",
"responses": {
"200": {
"description": "successful operation",
"content": {
"application/json": {
"schema": {
"type": "map",
"additionalProperties": {
"type": "integer",
"format": "int32"
}
}
}
}
}
},
"security": [
{
"api_key": []
}
]
}
},
"/store/order": {
"post": {
"tags": [
"store"
],
"summary": "Place an order for a pet",
"description": "Place a new order in the store",
"operationId": "placeOrder",
"requestBody": {
"description": "order placed for purchasing the pet",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Order"
}
},
"application/xml": {
"schema": {
"$ref": "#/components/schemas/Order"
}
},
"application/x-www-form-urlencoded": {
"schema": {
"$ref": "#/components/schemas/Order"
}
}
},
"required": true
},
"responses": {
"200": {
"description": "successful operation",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Order"
}
}
}
},
"405": {
"description": "Invalid input",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
}
}
}
},
"/store/order/{orderId}": {
"get": {
"tags": [
"store"
],
"summary": "Find purchase order by ID",
"description": "For valid response try integer IDs with value <= 5 or > 10. Other values will generated exceptions",
"operationId": "getOrderById",
"parameters": [
{
"name": "orderId",
"in": "path",
"description": "ID of order that needs to be fetched",
"required": true,
"schema": {
"type": "integer",
"format": "int64"
}
}
],
"responses": {
"200": {
"description": "successful operation",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Order"
}
},
"application/xml": {
"schema": {
"$ref": "#/components/schemas/Order"
}
}
}
},
"400": {
"description": "Invalid ID supplied",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
},
"404": {
"description": "Order not found",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
}
}
},
"delete": {
"tags": [
"store"
],
"summary": "Delete purchase order by ID",
"description": "For valid response try integer IDs with value < 1000. Anything above 10000 will generate exception",
"operationId": "deleteOrder",
"parameters": [
{
"name": "orderId",
"in": "path",
"description": "ID of the order that needs to be deleted",
"required": true,
"schema": {
"type": "integer",
"format": "int64"
}
}
],
"responses": {
"400": {
"description": "Invalid ID supplied",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
},
"404": {
"description": "Order not found",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
}
}
}
},
"/user": {
"post": {
"tags": [
"user"
],
"summary": "Create user",
"description": "This can only be done by the logged in user.",
"operationId": "createUser",
"requestBody": {
"description": "Created user object",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/User"
}
},
"application/xml": {
"schema": {
"$ref": "#/components/schemas/User"
}
},
"application/x-www-form-urlencoded": {
"schema": {
"$ref": "#/components/schemas/User"
}
}
},
"required": true
},
"responses": {
"default": {
"description": "successful operation",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/User"
}
},
"application/xml": {
"schema": {
"$ref": "#/components/schemas/User"
}
}
}
}
}
},
"get": {
"tags": [
"user"
],
"summary": "List users",
"operationId": "listUsers",
"responses": {
"200": {
"description": "Successful operation",
"content": {
"application/json": {
"schema": {
"type": "array",
"items": {
"$ref": "#/components/schemas/User"
}
}
}
}
}
}
}
},
"/user/createWithArray": {
"post": {
"tags": [
"user"
],
"summary": "Creates list of users with given input array",
"description": "Creates list of users with given input array",
"operationId": "createUsersWithArrayInput",
"requestBody": {
"description": "List of user object",
"content": {
"application/json": {
"schema": {
"type": "array",
"items": {
"$ref": "#/components/schemas/User"
}
}
}
},
"required": true
},
"responses": {
"default": {
"description": "successful operation",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/User"
}
},
"application/xml": {
"schema": {
"$ref": "#/components/schemas/User"
}
}
}
}
}
}
},
"/user/createWithList": {
"post": {
"tags": [
"user"
],
"summary": "Creates list of users with given input list",
"description": "Creates list of users with given input list",
"operationId": "createUsersWithListInput",
"requestBody": {
"description": "List of user object",
"content": {
"application/json": {
"schema": {
"type": "array",
"items": {
"$ref": "#/components/schemas/User"
}
}
}
},
"required": true
},
"responses": {
"default": {
"description": "successful operation",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/User"
}
},
"application/xml": {
"schema": {
"$ref": "#/components/schemas/User"
}
}
}
}
}
}
},
"/user/login": {
"get": {
"tags": [
"user"
],
"summary": "Logs user into the system",
"description": "",
"operationId": "loginUser",
"parameters": [
{
"name": "username",
"in": "query",
"description": "The user name for login",
"required": false,
"schema": {
"type": "string"
}
},
{
"name": "password",
"in": "query",
"description": "The password for login in clear text",
"required": false,
"schema": {
"type": "string"
}
}
],
"responses": {
"200": {
"description": "successful operation",
"headers": {
"X-Rate-Limit": {
"description": "calls per hour allowed by the user",
"schema": {
"type": "integer",
"format": "int32"
}
},
"X-Expires-After": {
"description": "date in UTC when token expires",
"schema": {
"type": "string",
"format": "date-time"
}
}
},
"content": {
"application/json": {
"schema": {
"type": "string"
}
},
"application/xml": {
"schema": {
"type": "string"
}
}
}
},
"400": {
"description": "Invalid username/password supplied",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
}
}
}
},
"/user/logout": {
"get": {
"tags": [
"user"
],
"summary": "Logs out current logged in user session",
"description": "",
"operationId": "logoutUser",
"responses": {
"default": {
"description": "successful operation",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
}
}
}
},
"/user/{username}": {
"get": {
"tags": [
"user"
],
"summary": "Get user by user name",
"description": "",
"operationId": "getUserByName",
"parameters": [
{
"name": "username",
"in": "path",
"description": "The name that needs to be fetched. Use user1 for testing. ",
"required": true,
"schema": {
"type": "string"
}
}
],
"responses": {
"200": {
"description": "successful operation",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/User"
}
},
"application/xml": {
"schema": {
"$ref": "#/components/schemas/User"
}
}
}
},
"400": {
"description": "Invalid username supplied",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
},
"404": {
"description": "User not found",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
}
}
},
"put": {
"tags": [
"user"
],
"summary": "Update user",
"description": "This can only be done by the logged in user.",
"operationId": "updateUser",
"parameters": [
{
"name": "username",
"in": "path",
"description": "name that need to be updated",
"required": true,
"schema": {
"type": "string"
}
},
{
"name": "body",
"in": "body",
"description": "Update an existent user in the store",
"required": false,
"schema": {
"$ref": "#/components/schemas/User"
}
}
],
"responses": {
"default": {
"description": "successful operation",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
},
"400": {
"description": "Invalid user supplied",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
},
"404": {
"description": "User not found",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
}
}
},
"delete": {
"tags": [
"user"
],
"summary": "Delete user",
"description": "This can only be done by the logged in user.",
"operationId": "deleteUser",
"parameters": [
{
"name": "username",
"in": "path",
"description": "The name that needs to be deleted",
"required": true,
"schema": {
"type": "string"
}
}
],
"responses": {
"400": {
"description": "Invalid username supplied",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
},
"404": {
"description": "User not found",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
}
}
}
}
},
"components": {
"schemas": {
"Order": {
"type": "object",
"properties": {
"id": {
"type": "integer",
"format": "int64"
},
"petId": {
"type": "integer",
"format": "int64"
},
"quantity": {
"type": "integer",
"format": "int32"
},
"shipDate": {
"type": "string",
"format": "date-time"
},
"status": {
"type": "string",
"description": "Order Status",
"enum": [
"placed",
"approved",
"delivered"
]
},
"complete": {
"type": "boolean",
"default": false
}
},
"xml": {
"name": "Order"
}
},
"Category": {
"type": "object",
"properties": {
"id": {
"type": "integer",
"format": "int64"
},
"name": {
"type": "string",
"xml": {
"name": "name"
}
}
},
"xml": {
"name": "Category"
}
},
"User": {
"type": "object",
"properties": {
"id": {
"type": "integer",
"format": "int64"
},
"username": {
"type": "string",
"xml": {
"name": "username"
}
},
"firstName": {
"type": "string",
"xml": {
"name": "firstName"
}
},
"lastName": {
"type": "string",
"xml": {
"name": "lastName"
}
},
"email": {
"type": "string",
"xml": {
"name": "email"
}
},
"password": {
"type": "string",
"xml": {
"name": "password"
}
},
"phone": {
"type": "string",
"xml": {
"name": "phone"
}
},
"userStatus": {
"type": "integer",
"format": "int32",
"description": "User Status"
}
},
"xml": {
"name": "User"
}
},
"Customer": {
"type": "object",
"properties": {
"id": {
"type": "integer",
"format": "int64"
},
"username": {
"type": "string",
"xml": {
"name": "username"
}
},
"address": {
"type": "array",
"xml": {
"name": "addresses"
},
"items": {
"$ref": "#/components/schemas/Address"
}
}
},
"xml": {
"name": "Customer"
}
},
"Address": {
"type": "object",
"properties": {
"street": {
"type": "string",
"xml": {
"name": "street"
}
},
"city": {
"type": "string",
"xml": {
"name": "city"
}
},
"state": {
"type": "string",
"xml": {
"name": "state"
}
},
"zip": {
"type": "string",
"xml": {
"name": "zip"
}
}
},
"xml": {
"name": "Address"
}
},
"Pet": {
"type": "object",
"required": [
"name",
"photoUrls"
],
"properties": {
"id": {
"type": "integer",
"format": "int64",
"xml": {
"name": "id"
}
},
"category": {
"$ref": "#/components/schemas/Category"
},
"name": {
"type": "string",
"xml": {
"name": "name"
}
},
"photoUrls": {
"type": "array",
"xml": {
"name": "photoUrl",
"wrapped": true
},
"items": {
"type": "string"
}
},
"tags": {
"type": "array",
"xml": {
"name": "tag",
"wrapped": true
},
"items": {
"$ref": "#/components/schemas/Tag"
}
},
"status": {
"type": "string",
"description": "pet status in the store",
"enum": [
"available",
"pending",
"sold"
],
"xml": {
"name": "status"
}
}
},
"xml": {
"name": "Pet"
}
},
"Tag": {
"type": "object",
"properties": {
"id": {
"type": "integer",
"format": "int64"
},
"name": {
"type": "string",
"xml": {
"name": "name"
}
}
},
"xml": {
"name": "Tag"
}
},
"ApiResponse": {
"type": "object",
"properties": {
"code": {
"type": "integer",
"format": "int32"
},
"type": {
"type": "string"
},
"message": {
"type": "string"
}
},
"xml": {
"name": "##default"
}
},
"Error": {
"type": "object",
"required": [
"message"
],
"properties": {
"code": {
"type": "integer",
"format": "int32"
},
"message": {
"type": "string"
},
"fields": {
"type": "string"
}
}
}
},
"requestBodies": {
"Pet": {
"description": "Pet object that needs to be added to the store",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Pet"
}
},
"application/xml": {
"schema": {
"$ref": "#/components/schemas/Pet"
}
}
}
},
"UserArray": {
"description": "List of user object",
"content": {
"application/json": {
"schema": {
"type": "array",
"items": {
"$ref": "#/components/schemas/User"
}
}
}
}
}
},
"securitySchemes": {
"petstore_auth": {
"type": "oauth2",
"flows": {
"implicit": {
"authorizationUrl": "http://petstore.swagger.io/oauth/dialog",
"scopes": {
"write:pets": "modify pets in your account",
"read:pets": "read your pets"
}
}
}
},
"api_key": {
"type": "apiKey",
"name": "api_key",
"in": "header"
}
}
}
}
```
--------------------------------------------------------------------------------
/examples/getzep.swagger.json:
--------------------------------------------------------------------------------
```json
{
"swagger": "2.0",
"info": {
"title": "Zep Memory API (V2)",
"description": "OpenAPI specification for Zep API V2, covering Memory, Graph, Group, and User endpoints",
"version": "2.0"
},
"host": "api.getzep.com",
"basePath": "/api/v2",
"schemes": ["https"],
"securityDefinitions": {
"ApiKeyAuth": {
"type": "apiKey",
"in": "header",
"name": "Authorization",
"description": "Header authentication in the format: Api-Key <token>"
}
},
"security": [{"ApiKeyAuth": []}],
"paths": {
"/sessions": {
"post": {
"summary": "Add Session",
"description": "Creates a new session.",
"consumes": ["application/json"],
"produces": ["application/json"],
"parameters": [
{
"in": "body",
"name": "body",
"description": "Request body",
"required": true,
"schema": {
"type": "object",
"required": ["session_id", "user_id"],
"properties": {
"session_id": {"type": "string", "description": "Unique identifier of the session"},
"user_id": {"type": "string", "description": "Unique identifier of the user associated with the session"},
"fact_rating_instruction": {
"type": "object",
"description": "Deprecated",
"properties": {
"instruction": {"type": "string"},
"examples": {
"type": "object",
"properties": {
"high": {"type": "string"},
"low": {"type": "string"},
"medium": {"type": "string"}
}
}
}
},
"metadata": {
"type": "object",
"description": "Deprecated",
"additionalProperties": {}
}
}
}
}
],
"responses": {
"201": {
"description": "Created",
"schema": {"$ref": "#/definitions/Session"}
},
"400": {"description": "Memory Add Session Request Bad Request Error"},
"500": {"description": "Memory Add Session Request Internal Server Error"}
}
}
},
"/sessions/{sessionId}": {
"get": {
"summary": "Get Session",
"description": "Retrieves a session by ID.",
"produces": ["application/json"],
"parameters": [
{
"in": "path",
"name": "sessionId",
"description": "Unique identifier of the session",
"required": true,
"type": "string"
}
],
"responses": {
"200": {
"description": "Retrieved",
"schema": {"$ref": "#/definitions/Session"}
},
"404": {"description": "Memory Get Session Request Not Found Error"},
"500": {"description": "Memory Get Session Request Internal Server Error"}
}
}
},
"/sessions-ordered": {
"get": {
"summary": "Get Sessions",
"description": "Retrieves all sessions with optional sorting and pagination.",
"produces": ["application/json"],
"parameters": [
{
"name": "page_number",
"in": "query",
"type": "integer",
"description": "Page number for pagination, starting from 1",
"required": false
},
{
"name": "page_size",
"in": "query",
"type": "integer",
"description": "Number of sessions per page",
"required": false
},
{
"name": "order_by",
"in": "query",
"type": "string",
"description": "Field to order results by: created_at, updated_at, user_id, session_id",
"required": false
},
{
"name": "asc",
"in": "query",
"type": "boolean",
"description": "Order direction: true for ascending, false for descending",
"required": false
}
],
"responses": {
"200": {
"description": "Retrieved",
"schema": {
"type": "object",
"properties": {
"response_count": {"type": "integer"},
"sessions": {"type": "array", "items": {"$ref": "#/definitions/Session"}},
"total_count": {"type": "integer"}
}
}
},
"400": {"description": "Memory List Sessions Request Bad Request Error"},
"500": {"description": "Memory List Sessions Request Internal Server Error"}
}
}
},
"/sessions/{sessionId}/memory": {
"get": {
"summary": "Get Session Memory",
"description": "Retrieves memory for a given session.",
"produces": ["application/json"],
"parameters": [
{
"in": "path",
"name": "sessionId",
"description": "ID of the session to retrieve memory for",
"required": true,
"type": "string"
},
{
"name": "lastn",
"in": "query",
"type": "integer",
"description": "Number of most recent memory entries to retrieve",
"required": false
},
{
"name": "minRating",
"in": "query",
"type": "number",
"format": "double",
"description": "Minimum rating to filter relevant facts",
"required": false
}
],
"responses": {
"200": {
"description": "Retrieved",
"schema": {"$ref": "#/definitions/Memory"}
},
"404": {"description": "Memory Get Request Not Found Error"},
"500": {"description": "Memory Get Request Internal Server Error"}
}
},
"post": {
"summary": "Add Memory to Session",
"description": "Adds memory to a specified session.",
"consumes": ["application/json"],
"produces": ["application/json"],
"parameters": [
{
"in": "path",
"name": "sessionId",
"description": "ID of the session to add memory to",
"required": true,
"type": "string"
},
{
"in": "body",
"name": "body",
"description": "Request body",
"required": true,
"schema": {
"type": "object",
"required": ["messages"],
"properties": {
"messages": {
"type": "array",
"items": {
"type": "object",
"required": ["content", "role_type"],
"properties": {
"content": {"type": "string", "description": "Message content"},
"role_type": {
"type": "string",
"description": "Role type (e.g., user, system)",
"enum": ["assistant", "user", "system", "function", "norole"]
},
"metadata": {
"type": "object",
"description": "Message metadata",
"additionalProperties": {}
},
"role": {"type": "string", "description": "Custom role of the sender (e.g., john, sales_agent)"},
"token_count": {"type": "integer", "description": "Deprecated"}
}
}
},
"fact_instruction": {"type": "string", "description": "Deprecated"},
"return_context": {"type": "boolean", "description": "Optionally return memory context for recent messages"},
"summary_instruction": {"type": "string", "description": "Deprecated"}
}
}
}
],
"responses": {
"200": {
"description": "Successful",
"schema": {
"type": "object",
"properties": {"context": {"type": "string"}}
}
},
"500": {"description": "Memory Add Request Internal Server Error"}
}
},
"delete": {
"summary": "Delete Session",
"description": "Deletes a session.",
"produces": ["application/json"],
"parameters": [
{
"in": "path",
"name": "sessionId",
"description": "ID of the session to delete",
"required": true,
"type": "string"
}
],
"responses": {
"200": {
"description": "Deleted",
"schema": {
"type": "object",
"properties": {"message": {"type": "string"}}
}
},
"404": {"description": "Memory Delete Request Not Found Error"},
"500": {"description": "Memory Delete Request Internal Server Error"}
}
}
},
"/sessions/{sessionId}/messages": {
"get": {
"summary": "Get Messages for Session",
"description": "Retrieves messages for a session.",
"produces": ["application/json"],
"parameters": [
{
"in": "path",
"name": "sessionId",
"description": "Session ID",
"required": true,
"type": "string"
},
{
"name": "limit",
"in": "query",
"type": "integer",
"description": "Limit the number of results returned",
"required": false
},
{
"name": "cursor",
"in": "query",
"type": "integer",
"description": "Cursor for pagination",
"required": false
}
],
"responses": {
"200": {
"description": "Retrieved",
"schema": {
"type": "object",
"properties": {
"messages": {"type": "array", "items": {"$ref": "#/definitions/Message"}},
"row_count": {"type": "integer"},
"total_count": {"type": "integer"}
}
}
},
"404": {"description": "Memory Get Session Messages Request Not Found Error"},
"500": {"description": "Memory Get Session Messages Request Internal Server Error"}
}
}
},
"/sessions/{sessionId}/messages/classify": {
"post": {
"summary": "Classify Session",
"description": "Classifies a session.",
"consumes": ["application/json"],
"produces": ["application/json"],
"parameters": [
{
"in": "path",
"name": "sessionId",
"description": "Session ID",
"required": true,
"type": "string"
},
{
"in": "body",
"name": "body",
"description": "Request body",
"required": true,
"schema": {
"type": "object",
"required": ["classes", "name"],
"properties": {
"classes": {
"type": "array",
"items": {"type": "string", "description": "Classes for classification"}
},
"name": {"type": "string", "description": "Name of the classifier"},
"instruction": {"type": "string", "description": "Custom instruction for classification"},
"last_n": {"type": "integer", "description": "Number of session messages to consider, defaults to 4"},
"persist": {"type": "boolean", "description": "Deprecated, defaults to true"}
}
}
}
],
"responses": {
"200": {
"description": "Successful",
"schema": {
"type": "object",
"properties": {
"class": {"type": "string"},
"label": {"type": "string"}
}
}
},
"404": {"description": "Memory Classify Session Request Not Found Error"},
"500": {"description": "Memory Classify Session Request Internal Server Error"}
}
}
},
"/sessions/{sessionId}/messages/{messageUUID}": {
"get": {
"summary": "Get Message",
"description": "Retrieves a specific message from a session.",
"produces": ["application/json"],
"parameters": [
{
"in": "path",
"name": "sessionId",
"description": "Soon to be deprecated, not needed",
"required": true,
"type": "string"
},
{
"in": "path",
"name": "messageUUID",
"description": "UUID of the message",
"required": true,
"type": "string"
}
],
"responses": {
"200": {
"description": "Retrieved",
"schema": {"$ref": "#/definitions/Message"}
},
"404": {"description": "Memory Get Session Message Request Not Found Error"},
"500": {"description": "Memory Get Session Message Request Internal Server Error"}
}
}
},
"/graph": {
"post": {
"summary": "Add Data to Graph",
"description": "Adds data to the graph. Limits vary by subscription tier; see pricing page.",
"consumes": ["application/json"],
"produces": ["application/json"],
"parameters": [
{
"in": "body",
"name": "body",
"description": "Request body",
"required": true,
"schema": {
"type": "object",
"properties": {
"data": {"type": "string", "description": "Optional data to add"},
"group_id": {"type": "string", "description": "Optional group ID"},
"type": {
"type": "string",
"enum": ["text", "json", "message"],
"description": "Optional data type"
},
"user_id": {"type": "string", "description": "Optional user ID"}
}
}
}
],
"responses": {
"202": {
"description": "Accepted",
"schema": {
"type": "object",
"properties": {"message": {"type": "string"}}
}
},
"400": {"description": "Graph Add Request Bad Request Error"},
"500": {"description": "Graph Add Request Internal Server Error"}
}
}
},
"/graph/add-fact-triple": {
"post": {
"summary": "Add Fact Triple",
"description": "Adds a fact triple for a user or group.",
"consumes": ["application/json"],
"produces": ["application/json"],
"parameters": [
{
"in": "body",
"name": "body",
"description": "Request body",
"required": true,
"schema": {
"type": "object",
"required": ["fact", "fact_name", "target_node_name"],
"properties": {
"fact": {"type": "string", "maxLength": 50, "description": "Fact relating the nodes"},
"fact_name": {"type": "string", "minLength": 1, "maxLength": 50, "description": "Edge name in CAPS_SNAKE_CASE"},
"target_node_name": {"type": "string", "maxLength": 50, "description": "Target node name"},
"created_at": {"type": "string", "description": "Optional timestamp"},
"expired_at": {"type": "string", "description": "Optional expiry time"},
"fact_uuid": {"type": "string", "description": "Optional edge UUID"},
"group_id": {"type": "string", "description": "Optional group ID"},
"invalid_at": {"type": "string", "description": "Optional invalidation time"},
"source_node_name": {"type": "string", "maxLength": 50, "description": "Optional source node name"},
"source_node_summary": {"type": "string", "maxLength": 500, "description": "Optional source summary"},
"source_node_uuid": {"type": "string", "description": "Optional source UUID"},
"target_node_summary": {"type": "string", "maxLength": 500, "description": "Optional target summary"},
"target_node_uuid": {"type": "string", "description": "Optional target UUID"},
"user_id": {"type": "string", "description": "Optional user ID"},
"valid_at": {"type": "string", "description": "Optional valid-from time"}
}
}
}
],
"responses": {
"200": {
"description": "Successful",
"schema": {
"type": "object",
"properties": {
"edge": {"$ref": "#/definitions/Edge"},
"source_node": {"$ref": "#/definitions/Node"},
"target_node": {"$ref": "#/definitions/Node"}
}
}
},
"400": {"description": "Graph Add Fact Triple Request Bad Request Error"},
"500": {"description": "Graph Add Fact Triple Request Internal Server Error"}
}
}
},
"/graph/search": {
"post": {
"summary": "Search Graph",
"description": "Performs a graph search query.",
"consumes": ["application/json"],
"produces": ["application/json"],
"parameters": [
{
"in": "body",
"name": "body",
"description": "Request body",
"required": true,
"schema": {
"type": "object",
"required": ["query"],
"properties": {
"query": {"type": "string", "description": "Search query string"},
"center_node_uuid": {"type": "string", "description": "Optional node to rerank around"},
"group_id": {"type": "string", "description": "Optional group ID"},
"limit": {"type": "integer", "default": 10, "maximum": 50, "description": "Maximum number of facts to retrieve"},
"min_score": {"type": "number", "format": "double", "description": "Deprecated"},
"mmr_lambda": {"type": "number", "format": "double", "description": "Maximal marginal relevance weighting"},
"reranker": {
"type": "string",
"enum": ["rrf", "mmr", "node_distance", "episode_mentions", "cross_encoder"],
"default": "rrf",
"description": "Reranker type"
},
"scope": {
"type": "string",
"enum": ["edges", "nodes"],
"default": "edges",
"description": "Search scope"
},
"search_filters": {"type": "object", "description": "Optional search filters"},
"user_id": {"type": "string", "description": "Optional user ID"}
}
}
}
],
"responses": {
"200": {
"description": "Successful",
"schema": {
"type": "object",
"properties": {
"edges": {"type": "array", "items": {"$ref": "#/definitions/Edge"}},
"nodes": {"type": "array", "items": {"$ref": "#/definitions/Node"}}
}
}
},
"400": {"description": "Graph Search Request Bad Request Error"},
"500": {"description": "Graph Search Request Internal Server Error"}
}
}
},
"/groups": {
"post": {
"summary": "Create Group",
"description": "Creates a new group.",
"consumes": ["application/json"],
"produces": ["application/json"],
"parameters": [
{
"in": "body",
"name": "body",
"description": "Request body",
"required": true,
"schema": {
"type": "object",
"required": ["group_id"],
"properties": {
"group_id": {"type": "string", "description": "Unique group identifier"},
"description": {"type": "string", "description": "Optional description"},
"fact_rating_instruction": {
"type": "object",
"properties": {
"instruction": {"type": "string"},
"examples": {
"type": "object",
"properties": {
"high": {"type": "string"},
"low": {"type": "string"},
"medium": {"type": "string"}
}
}
}
},
"name": {"type": "string", "description": "Optional group name"}
}
}
}
],
"responses": {
"201": {
"description": "Created",
"schema": {"$ref": "#/definitions/Group"}
},
"400": {"description": "Group Add Request Bad Request Error"},
"500": {"description": "Group Add Request Internal Server Error"}
}
}
},
"/groups-ordered": {
"get": {
"summary": "Get All Groups",
"description": "Retrieves all groups with pagination.",
"produces": ["application/json"],
"parameters": [
{
"name": "pageNumber",
"in": "query",
"type": "integer",
"description": "Page number, starting from 1",
"required": false
},
{
"name": "pageSize",
"in": "query",
"type": "integer",
"description": "Number of groups per page",
"required": false
}
],
"responses": {
"200": {
"description": "Retrieved",
"schema": {
"type": "object",
"properties": {
"groups": {"type": "array", "items": {"$ref": "#/definitions/Group"}},
"row_count": {"type": "integer"},
"total_count": {"type": "integer"}
}
}
},
"400": {"description": "Get Groups Ordered Request Bad Request Error"},
"500": {"description": "Get Groups Ordered Request Internal Server Error"}
}
}
},
"/groups/{groupId}": {
"get": {
"summary": "Get Group",
"description": "Retrieves a group by ID.",
"produces": ["application/json"],
"parameters": [
{
"in": "path",
"name": "groupId",
"type": "string",
"required": true,
"description": "Group ID"
}
],
"responses": {
"200": {
"description": "Retrieved",
"schema": {"$ref": "#/definitions/Group"}
},
"404": {"description": "Get Groups Group ID Request Not Found Error"},
"500": {"description": "Get Groups Group ID Request Internal Server Error"}
}
},
"delete": {
"summary": "Delete Group",
"description": "Deletes a group.",
"produces": ["application/json"],
"parameters": [
{
"in": "path",
"name": "groupId",
"type": "string",
"required": true,
"description": "Group ID to delete"
}
],
"responses": {
"200": {
"description": "Deleted",
"schema": {
"type": "object",
"properties": {"message": {"type": "string"}}
}
},
"400": {"description": "Group Delete Request Bad Request Error"},
"404": {"description": "Group Delete Request Not Found Error"},
"500": {"description": "Group Delete Request Internal Server Error"}
}
},
"patch": {
"summary": "Update Group",
"description": "Updates group information.",
"consumes": ["application/json"],
"produces": ["application/json"],
"parameters": [
{
"in": "path",
"name": "groupId",
"type": "string",
"required": true,
"description": "Group ID to update"
},
{
"in": "body",
"name": "body",
"description": "Update body",
"required": true,
"schema": {
"type": "object",
"properties": {
"description": {"type": "string", "description": "Optional new description"},
"fact_rating_instruction": {
"type": "object",
"properties": {
"instruction": {"type": "string"},
"examples": {
"type": "object",
"properties": {
"high": {"type": "string"},
"low": {"type": "string"},
"medium": {"type": "string"}
}
}
}
},
"name": {"type": "string", "description": "Optional new name"}
}
}
}
],
"responses": {
"201": {
"description": "Updated",
"schema": {"$ref": "#/definitions/Group"}
},
"400": {"description": "Group Update Request Bad Request Error"},
"404": {"description": "Group Update Request Not Found Error"},
"500": {"description": "Group Update Request Internal Server Error"}
}
}
},
"/users": {
"post": {
"summary": "Add User",
"description": "Creates a new user.",
"consumes": ["application/json"],
"produces": ["application/json"],
"parameters": [
{
"in": "body",
"name": "body",
"description": "Request body",
"required": true,
"schema": {
"type": "object",
"properties": {
"email": {"type": "string", "description": "Optional email address"},
"fact_rating_instruction": {
"type": "object",
"description": "Optional fact rating instruction",
"properties": {
"instruction": {"type": "string"},
"examples": {
"type": "object",
"properties": {
"high": {"type": "string"},
"low": {"type": "string"},
"medium": {"type": "string"}
}
}
}
},
"first_name": {"type": "string", "description": "Optional first name"},
"last_name": {"type": "string", "description": "Optional last name"},
"metadata": {
"type": "object",
"description": "Optional metadata",
"additionalProperties": {}
},
"user_id": {"type": "string", "description": "Optional user identifier"}
}
}
}
],
"responses": {
"201": {
"description": "Created",
"schema": {"$ref": "#/definitions/User"}
},
"400": {"description": "User Add Request Bad Request Error"},
"500": {"description": "User Add Request Internal Server Error"}
}
}
},
"/users-ordered": {
"get": {
"summary": "Get Users",
"description": "Retrieves all users with pagination.",
"produces": ["application/json"],
"parameters": [
{
"name": "pageNumber",
"in": "query",
"type": "integer",
"description": "Page number, starting from 1",
"required": false
},
{
"name": "pageSize",
"in": "query",
"type": "integer",
"description": "Number of users per page",
"required": false
}
],
"responses": {
"200": {
"description": "Retrieved",
"schema": {
"type": "object",
"properties": {
"row_count": {"type": "integer"},
"total_count": {"type": "integer"},
"users": {"type": "array", "items": {"$ref": "#/definitions/User"}}
}
}
},
"400": {"description": "User List Ordered Request Bad Request Error"},
"500": {"description": "User List Ordered Request Internal Server Error"}
}
}
},
"/users/{userId}": {
"get": {
"summary": "Get User",
"description": "Retrieves a user by ID.",
"produces": ["application/json"],
"parameters": [
{
"in": "path",
"name": "userId",
"type": "string",
"required": true,
"description": "User ID"
}
],
"responses": {
"200": {
"description": "Retrieved",
"schema": {"$ref": "#/definitions/User"}
},
"404": {"description": "User Get Request Not Found Error"},
"500": {"description": "User Get Request Internal Server Error"}
}
},
"delete": {
"summary": "Delete User",
"description": "Deletes a user.",
"produces": ["application/json"],
"parameters": [
{
"in": "path",
"name": "userId",
"type": "string",
"required": true,
"description": "User ID to delete"
}
],
"responses": {
"200": {
"description": "Deleted",
"schema": {
"type": "object",
"properties": {"message": {"type": "string"}}
}
},
"404": {"description": "User Delete Request Not Found Error"},
"500": {"description": "User Delete Request Internal Server Error"}
}
},
"patch": {
"summary": "Update User",
"description": "Updates user information.",
"consumes": ["application/json"],
"produces": ["application/json"],
"parameters": [
{
"in": "path",
"name": "userId",
"type": "string",
"required": true,
"description": "User ID to update"
},
{
"in": "body",
"name": "body",
"description": "Update body",
"required": true,
"schema": {
"type": "object",
"properties": {
"email": {"type": "string", "description": "Optional email address"},
"fact_rating_instruction": {
"type": "object",
"description": "Optional fact rating instruction",
"properties": {
"instruction": {"type": "string"},
"examples": {
"type": "object",
"properties": {
"high": {"type": "string"},
"low": {"type": "string"},
"medium": {"type": "string"}
}
}
}
},
"first_name": {"type": "string", "description": "Optional first name"},
"last_name": {"type": "string", "description": "Optional last name"},
"metadata": {
"type": "object",
"description": "Optional metadata",
"additionalProperties": {}
}
}
}
}
],
"responses": {
"200": {
"description": "Updated",
"schema": {"$ref": "#/definitions/User"}
},
"400": {"description": "User Update Request Bad Request Error"},
"404": {"description": "User Update Request Not Found Error"},
"500": {"description": "User Update Request Internal Server Error"}
}
}
},
"/users/{userId}/node": {
"get": {
"summary": "Get User Node",
"description": "Retrieves a user’s graph node.",
"produces": ["application/json"],
"parameters": [
{
"in": "path",
"name": "userId",
"type": "string",
"required": true,
"description": "User ID for the node"
}
],
"responses": {
"200": {
"description": "Retrieved",
"schema": {
"type": "object",
"properties": {"node": {"$ref": "#/definitions/Node"}}
}
},
"404": {"description": "User Get Node Request Not Found Error"},
"500": {"description": "User Get Node Request Internal Server Error"}
}
}
},
"/users/{userId}/sessions": {
"get": {
"summary": "Get User Sessions",
"description": "Retrieves all sessions for a user.",
"produces": ["application/json"],
"parameters": [
{
"in": "path",
"name": "userId",
"type": "string",
"required": true,
"description": "User ID for sessions"
}
],
"responses": {
"200": {
"description": "Retrieved",
"schema": {"type": "array", "items": {"$ref": "#/definitions/Session"}}
},
"500": {"description": "User Get Sessions Request Internal Server Error"}
}
}
}
},
"definitions": {
"Session": {
"type": "object",
"properties": {
"classifications": {
"type": "object",
"additionalProperties": {"type": "string"}
},
"created_at": {"type": "string"},
"deleted_at": {"type": "string"},
"ended_at": {"type": "string"},
"fact_rating_instruction": {
"type": "object",
"description": "Deprecated",
"properties": {
"instruction": {"type": "string"},
"examples": {
"type": "object",
"properties": {
"high": {"type": "string"},
"low": {"type": "string"},
"medium": {"type": "string"}
}
}
}
},
"facts": {
"type": "array",
"items": {"type": "string"},
"description": "Deprecated"
},
"id": {"type": "integer"},
"metadata": {
"type": "object",
"description": "Deprecated",
"additionalProperties": {}
},
"project_uuid": {"type": "string"},
"session_id": {"type": "string"},
"updated_at": {"type": "string", "description": "Deprecated"},
"user_id": {"type": "string"},
"uuid": {"type": "string"}
}
},
"Message": {
"type": "object",
"properties": {
"content": {"type": "string", "description": "Message content"},
"role_type": {
"type": "string",
"description": "Role type (e.g., user, system)",
"enum": ["assistant", "user", "system", "function", "norole"]
},
"created_at": {"type": "string", "description": "Timestamp of message creation"},
"metadata": {
"type": "object",
"description": "Message metadata",
"additionalProperties": {}
},
"role": {"type": "string", "description": "Custom role of the sender (e.g., john, sales_agent)"},
"token_count": {"type": "integer", "description": "Deprecated"},
"updated_at": {"type": "string", "description": "Deprecated"},
"uuid": {"type": "string", "description": "Unique identifier of the message"}
}
},
"Memory": {
"type": "object",
"properties": {
"context": {"type": "string"},
"facts": {
"type": "array",
"items": {"type": "string"},
"description": "Deprecated: Use relevant_facts instead"
},
"messages": {
"type": "array",
"items": {"$ref": "#/definitions/Message"},
"description": "List of message objects; only last_n messages returned"
},
"metadata": {
"type": "object",
"description": "Deprecated",
"additionalProperties": {}
},
"relevant_facts": {
"type": "array",
"items": {
"type": "object",
"properties": {
"content": {"type": "string"},
"created_at": {"type": "string"},
"fact": {"type": "string"},
"uuid": {"type": "string"},
"expired_at": {"type": "string"},
"invalid_at": {"type": "string"},
"name": {"type": "string"},
"rating": {"type": "number", "format": "double"},
"source_node_name": {"type": "string"},
"target_node_name": {"type": "string"},
"valid_at": {"type": "string"}
}
},
"description": "Most relevant facts to recent session messages"
},
"summary": {
"type": "object",
"description": "Deprecated: Use context string instead",
"properties": {
"content": {"type": "string"},
"created_at": {"type": "string"},
"metadata": {"type": "object", "additionalProperties": {}},
"related_message_uuids": {"type": "array", "items": {"type": "string"}},
"token_count": {"type": "integer"},
"uuid": {"type": "string"}
}
}
}
},
"ClassificationResult": {
"type": "object",
"properties": {
"class": {"type": "string"},
"label": {"type": "string"}
}
},
"Edge": {
"type": "object",
"properties": {
"created_at": {"type": "string"},
"fact": {"type": "string"},
"name": {"type": "string"},
"source_node_uuid": {"type": "string"},
"target_node_uuid": {"type": "string"},
"uuid": {"type": "string"},
"episodes": {"type": "array", "items": {"type": "string"}},
"expired_at": {"type": "string"},
"invalid_at": {"type": "string"},
"valid_at": {"type": "string"}
}
},
"Node": {
"type": "object",
"properties": {
"created_at": {"type": "string"},
"name": {"type": "string"},
"summary": {"type": "string"},
"uuid": {"type": "string"},
"attributes": {"type": "object", "additionalProperties": {"type": "string"}},
"labels": {"type": "array", "items": {"type": "string"}}
}
},
"Group": {
"type": "object",
"properties": {
"created_at": {"type": "string"},
"description": {"type": "string"},
"external_id": {"type": "string", "description": "Deprecated"},
"fact_rating_instruction": {
"type": "object",
"properties": {
"instruction": {"type": "string"},
"examples": {
"type": "object",
"properties": {
"high": {"type": "string"},
"low": {"type": "string"},
"medium": {"type": "string"}
}
}
}
},
"group_id": {"type": "string"},
"id": {"type": "integer"},
"name": {"type": "string"},
"project_uuid": {"type": "string"},
"uuid": {"type": "string"}
}
},
"User": {
"type": "object",
"properties": {
"created_at": {"type": "string"},
"deleted_at": {"type": "string"},
"email": {"type": "string"},
"fact_rating_instruction": {
"type": "object",
"properties": {
"instruction": {"type": "string"},
"examples": {
"type": "object",
"properties": {
"high": {"type": "string"},
"low": {"type": "string"},
"medium": {"type": "string"}
}
}
}
},
"first_name": {"type": "string"},
"id": {"type": "integer"},
"last_name": {"type": "string"},
"metadata": {
"type": "object",
"description": "Deprecated",
"additionalProperties": {}
},
"project_uuid": {"type": "string"},
"session_count": {"type": "integer", "description": "Deprecated"},
"updated_at": {"type": "string", "description": "Deprecated"},
"user_id": {"type": "string"},
"uuid": {"type": "string"}
}
}
}
}
```