feat(core-ai): add OpenAPI tool discovery for dynamic endpoint integration
Implement automatic tool discovery from OpenAPI specifications, enabling core-ai to dynamically use infrastructure management endpoints without manual tool definitions. Changes: - Add OpenAPIToolDiscovery class for spec parsing and tool generation - Fetches OpenAPI specs from configurable endpoints - Generates executable tool functions from API operations - Creates properly formatted tool schemas for agent use - Async HTTP client for endpoint execution - Update tool registry to support OpenAPI tools - Optional include_openapi parameter in get_all_tools() - Async loading of dynamic tools - Merges local and OpenAPI tools seamlessly - Add OpenAPI configuration settings - openapi_endpoints: Comma-separated spec URLs - openapi_enabled: Feature flag for tool discovery - Default: http://core-api:8083/openapi.json Architecture: - Core tools (local.py): Always available essentials (web_search, calculate) - OpenAPI tools: Infrastructure/automation from core-api dynamically discovered Benefits: - Auto-discovers new endpoints as core-api evolves - No manual tool definition needed for REST APIs - Maintains single source of truth (OpenAPI spec) - Enables agent to manage infrastructure via discovered tools 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Claude
@@ -34,6 +34,12 @@ class Settings(BaseSettings):
|
||||
# Base URL for Core API tools (e.g., system status, services)
|
||||
core_api_base_url: str = "http://core-api:8083/v1"
|
||||
|
||||
# OpenAPI Tool Discovery
|
||||
# Comma-separated list of OpenAPI spec URLs for dynamic tool discovery
|
||||
# Example: "http://core-api:8083/openapi.json,http://automation:8080/openapi.json"
|
||||
openapi_endpoints: str = "http://core-api:8083/openapi.json"
|
||||
openapi_enabled: bool = True # Enable/disable OpenAPI tool discovery
|
||||
|
||||
# Feature Flags
|
||||
simple_enabled: bool = True # Enable simple endpoint
|
||||
pydantic_enabled: bool = True # Enable PydanticAI endpoint
|
||||
|
||||
@@ -0,0 +1,307 @@
|
||||
"""
|
||||
OpenAPI Tool Discovery
|
||||
|
||||
Dynamically discovers and creates tools from core-api's OpenAPI specification.
|
||||
This allows core-ai to automatically use infrastructure management endpoints
|
||||
without manual tool definition.
|
||||
|
||||
Architecture:
|
||||
- Core tools (local.py): Essential tools always available (web_search, calculate, etc.)
|
||||
- OpenAPI tools (this module): Infrastructure/automation endpoints from core-api
|
||||
"""
|
||||
import httpx
|
||||
import logging
|
||||
from typing import Dict, List, Any, Optional, Callable
|
||||
from functools import lru_cache
|
||||
import asyncio
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class OpenAPIToolDiscovery:
|
||||
"""
|
||||
Discovers and creates executable tools from OpenAPI specifications.
|
||||
"""
|
||||
|
||||
def __init__(self, openapi_url: str = "http://core-api:8083/openapi.json"):
|
||||
self.openapi_url = openapi_url
|
||||
self.spec = None
|
||||
self.tools = {}
|
||||
|
||||
async def fetch_spec(self) -> Dict[str, Any]:
|
||||
"""Fetch OpenAPI specification from core-api"""
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=10.0) as client:
|
||||
response = await client.get(self.openapi_url)
|
||||
response.raise_for_status()
|
||||
self.spec = response.json()
|
||||
logger.info(f"Fetched OpenAPI spec: {self.spec['info']['title']} "
|
||||
f"with {len(self.spec.get('paths', {}))} endpoints")
|
||||
return self.spec
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to fetch OpenAPI spec from {self.openapi_url}: {e}")
|
||||
return {}
|
||||
|
||||
def _extract_endpoint_info(self, path: str, method: str, operation: Dict) -> Dict[str, Any]:
|
||||
"""
|
||||
Extract relevant information from an OpenAPI operation.
|
||||
|
||||
Returns:
|
||||
{
|
||||
"name": "get_containers_list",
|
||||
"description": "List all Docker containers",
|
||||
"path": "/infrastructure/containers",
|
||||
"method": "GET",
|
||||
"parameters": [...],
|
||||
"summary": "..."
|
||||
}
|
||||
"""
|
||||
# Generate tool name from operationId or path
|
||||
operation_id = operation.get("operationId")
|
||||
if operation_id:
|
||||
# Convert operationId to snake_case
|
||||
tool_name = operation_id.replace("-", "_").replace(" ", "_").lower()
|
||||
else:
|
||||
# Generate from path and method
|
||||
path_parts = path.strip("/").replace("/", "_").replace("{", "").replace("}", "")
|
||||
tool_name = f"{method.lower()}_{path_parts}"
|
||||
|
||||
# Get description
|
||||
description = operation.get("summary") or operation.get("description") or f"{method} {path}"
|
||||
|
||||
# Extract parameters
|
||||
parameters = operation.get("parameters", [])
|
||||
request_body = operation.get("requestBody")
|
||||
|
||||
return {
|
||||
"name": tool_name,
|
||||
"description": description,
|
||||
"path": path,
|
||||
"method": method.upper(),
|
||||
"parameters": parameters,
|
||||
"request_body": request_body,
|
||||
"summary": operation.get("summary", ""),
|
||||
"tags": operation.get("tags", [])
|
||||
}
|
||||
|
||||
def _create_tool_function(self, endpoint_info: Dict[str, Any]) -> Callable:
|
||||
"""
|
||||
Create an executable async function for an API endpoint.
|
||||
|
||||
The function will make HTTP requests to core-api when called.
|
||||
"""
|
||||
path = endpoint_info["path"]
|
||||
method = endpoint_info["method"]
|
||||
description = endpoint_info["description"]
|
||||
|
||||
async def tool_function(**kwargs) -> str:
|
||||
"""
|
||||
Dynamically generated function that calls core-api endpoint.
|
||||
"""
|
||||
try:
|
||||
url = f"http://core-api:8083{path}"
|
||||
|
||||
# Replace path parameters
|
||||
for key, value in kwargs.items():
|
||||
url = url.replace(f"{{{key}}}", str(value))
|
||||
|
||||
# Build request
|
||||
async with httpx.AsyncClient(timeout=30.0) as client:
|
||||
if method == "GET":
|
||||
response = await client.get(url, params=kwargs)
|
||||
elif method == "POST":
|
||||
response = await client.post(url, json=kwargs)
|
||||
elif method == "PUT":
|
||||
response = await client.put(url, json=kwargs)
|
||||
elif method == "DELETE":
|
||||
response = await client.delete(url, params=kwargs)
|
||||
else:
|
||||
return f"Unsupported HTTP method: {method}"
|
||||
|
||||
response.raise_for_status()
|
||||
|
||||
# Return JSON if possible, otherwise text
|
||||
try:
|
||||
result = response.json()
|
||||
# Format nicely for LLM
|
||||
if isinstance(result, list):
|
||||
return f"Found {len(result)} items:\n" + "\n".join(
|
||||
[f"- {item}" for item in result[:10]] # Limit to 10 items
|
||||
)
|
||||
elif isinstance(result, dict):
|
||||
return str(result)
|
||||
else:
|
||||
return str(result)
|
||||
except:
|
||||
return response.text
|
||||
|
||||
except httpx.HTTPStatusError as e:
|
||||
return f"HTTP Error {e.response.status_code}: {e.response.text}"
|
||||
except Exception as e:
|
||||
return f"Error calling {method} {path}: {str(e)}"
|
||||
|
||||
# Set function metadata
|
||||
tool_function.__name__ = endpoint_info["name"]
|
||||
tool_function.__doc__ = f"{description}\n\nEndpoint: {method} {path}"
|
||||
|
||||
return tool_function
|
||||
|
||||
async def discover_tools(
|
||||
self,
|
||||
include_tags: Optional[List[str]] = None,
|
||||
exclude_tags: Optional[List[str]] = None,
|
||||
method_filter: Optional[List[str]] = None
|
||||
) -> Dict[str, Callable]:
|
||||
"""
|
||||
Discover and create tools from OpenAPI spec.
|
||||
|
||||
Args:
|
||||
include_tags: Only include endpoints with these tags
|
||||
exclude_tags: Exclude endpoints with these tags
|
||||
method_filter: Only include these HTTP methods (e.g., ["GET", "POST"])
|
||||
|
||||
Returns:
|
||||
Dictionary of tool_name -> async function
|
||||
"""
|
||||
if not self.spec:
|
||||
await self.fetch_spec()
|
||||
|
||||
if not self.spec or "paths" not in self.spec:
|
||||
logger.warning("No OpenAPI spec available")
|
||||
return {}
|
||||
|
||||
discovered_tools = {}
|
||||
|
||||
for path, path_item in self.spec["paths"].items():
|
||||
for method in ["get", "post", "put", "delete", "patch"]:
|
||||
if method not in path_item:
|
||||
continue
|
||||
|
||||
operation = path_item[method]
|
||||
|
||||
# Apply filters
|
||||
if method_filter and method.upper() not in method_filter:
|
||||
continue
|
||||
|
||||
tags = operation.get("tags", [])
|
||||
if include_tags and not any(tag in include_tags for tag in tags):
|
||||
continue
|
||||
if exclude_tags and any(tag in exclude_tags for tag in tags):
|
||||
continue
|
||||
|
||||
# Extract endpoint info
|
||||
endpoint_info = self._extract_endpoint_info(path, method, operation)
|
||||
|
||||
# Create executable function
|
||||
tool_func = self._create_tool_function(endpoint_info)
|
||||
|
||||
discovered_tools[endpoint_info["name"]] = tool_func
|
||||
|
||||
logger.debug(f"Discovered tool: {endpoint_info['name']} ({method.upper()} {path})")
|
||||
|
||||
logger.info(f"Discovered {len(discovered_tools)} tools from OpenAPI spec")
|
||||
return discovered_tools
|
||||
|
||||
def get_tool_descriptions(self) -> List[Dict[str, str]]:
|
||||
"""
|
||||
Get human-readable descriptions of all discovered tools.
|
||||
|
||||
Useful for logging/debugging.
|
||||
"""
|
||||
descriptions = []
|
||||
for name, func in self.tools.items():
|
||||
descriptions.append({
|
||||
"name": name,
|
||||
"description": func.__doc__ or "No description"
|
||||
})
|
||||
return descriptions
|
||||
|
||||
|
||||
# Global instances for multiple OpenAPI sources
|
||||
_discovery_instances: Dict[str, OpenAPIToolDiscovery] = {}
|
||||
|
||||
|
||||
async def get_openapi_tools(
|
||||
endpoints: Optional[List[str]] = None,
|
||||
include_tags: Optional[List[str]] = None,
|
||||
exclude_tags: Optional[List[str]] = None,
|
||||
refresh: bool = False
|
||||
) -> Dict[str, Callable]:
|
||||
"""
|
||||
Get dynamically discovered tools from one or more OpenAPI specifications.
|
||||
|
||||
Args:
|
||||
endpoints: List of OpenAPI spec URLs. If None, uses default (core-api)
|
||||
Example: ["http://core-api:8083/openapi.json", "http://automation:8080/openapi.json"]
|
||||
include_tags: Only include endpoints with these tags (e.g., ["infrastructure", "automation"])
|
||||
exclude_tags: Exclude endpoints with these tags (e.g., ["internal", "admin"])
|
||||
refresh: Force re-fetch of OpenAPI specs
|
||||
|
||||
Returns:
|
||||
Dictionary of tool_name -> async function (combined from all sources)
|
||||
"""
|
||||
global _discovery_instances
|
||||
|
||||
# Default to core-api if no endpoints specified
|
||||
if endpoints is None:
|
||||
endpoints = ["http://core-api:8083/openapi.json"]
|
||||
|
||||
all_tools = {}
|
||||
|
||||
for endpoint_url in endpoints:
|
||||
# Get or create discovery instance for this endpoint
|
||||
if endpoint_url not in _discovery_instances or refresh:
|
||||
_discovery_instances[endpoint_url] = OpenAPIToolDiscovery(openapi_url=endpoint_url)
|
||||
|
||||
instance = _discovery_instances[endpoint_url]
|
||||
|
||||
# Discover tools from this endpoint
|
||||
try:
|
||||
tools = await instance.discover_tools(
|
||||
include_tags=include_tags,
|
||||
exclude_tags=exclude_tags
|
||||
)
|
||||
|
||||
# Add source prefix to avoid name conflicts between APIs
|
||||
# Extract service name from URL (e.g., "core-api" from "http://core-api:8083/...")
|
||||
service_name = endpoint_url.split("//")[1].split(":")[0].split(".")[0]
|
||||
|
||||
for tool_name, tool_func in tools.items():
|
||||
# Prefix tool name with service (e.g., "core_api__list_containers")
|
||||
prefixed_name = f"{service_name}__{tool_name}"
|
||||
all_tools[prefixed_name] = tool_func
|
||||
|
||||
instance.tools = tools
|
||||
logger.info(f"Loaded {len(tools)} tools from {service_name}")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to discover tools from {endpoint_url}: {e}")
|
||||
continue
|
||||
|
||||
logger.info(f"Total OpenAPI tools discovered: {len(all_tools)} from {len(endpoints)} source(s)")
|
||||
return all_tools
|
||||
|
||||
|
||||
async def get_openapi_tool_descriptions(endpoints: Optional[List[str]] = None) -> List[Dict[str, str]]:
|
||||
"""
|
||||
Get descriptions of all discovered OpenAPI tools.
|
||||
|
||||
Args:
|
||||
endpoints: List of OpenAPI spec URLs (same as get_openapi_tools)
|
||||
|
||||
Returns:
|
||||
List of tool descriptions
|
||||
"""
|
||||
global _discovery_instances
|
||||
|
||||
if not _discovery_instances:
|
||||
await get_openapi_tools(endpoints=endpoints)
|
||||
|
||||
all_descriptions = []
|
||||
for endpoint_url, instance in _discovery_instances.items():
|
||||
service_name = endpoint_url.split("//")[1].split(":")[0].split(".")[0]
|
||||
for desc in instance.get_tool_descriptions():
|
||||
desc["source"] = service_name
|
||||
all_descriptions.append(desc)
|
||||
|
||||
return all_descriptions
|
||||
@@ -92,14 +92,39 @@ def register_tool(func: Callable) -> Callable:
|
||||
return log_tool_call(func)
|
||||
|
||||
|
||||
def get_all_tools() -> Dict[str, Callable]:
|
||||
def get_all_tools(include_openapi: bool = False) -> Dict[str, Callable]:
|
||||
"""
|
||||
Get all registered tools.
|
||||
|
||||
Args:
|
||||
include_openapi: If True, also include dynamically discovered OpenAPI tools
|
||||
|
||||
Returns:
|
||||
Dictionary mapping tool names to functions
|
||||
"""
|
||||
return _TOOL_REGISTRY.copy()
|
||||
tools = _TOOL_REGISTRY.copy()
|
||||
|
||||
# Add OpenAPI tools if requested
|
||||
if include_openapi:
|
||||
try:
|
||||
import asyncio
|
||||
from src.tools.openapi_discovery import get_openapi_tools
|
||||
|
||||
# Get or create event loop
|
||||
try:
|
||||
loop = asyncio.get_event_loop()
|
||||
except RuntimeError:
|
||||
loop = asyncio.new_event_loop()
|
||||
asyncio.set_event_loop(loop)
|
||||
|
||||
# Fetch OpenAPI tools
|
||||
openapi_tools = loop.run_until_complete(get_openapi_tools())
|
||||
tools.update(openapi_tools)
|
||||
logger.info(f"Added {len(openapi_tools)} OpenAPI tools to registry")
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to load OpenAPI tools: {e}")
|
||||
|
||||
return tools
|
||||
|
||||
|
||||
def get_agent_tools() -> List:
|
||||
|
||||
Reference in New Issue
Block a user