diff --git a/services/core-ai/src/config.py b/services/core-ai/src/config.py index a9d2fa8..b0077d7 100644 --- a/services/core-ai/src/config.py +++ b/services/core-ai/src/config.py @@ -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 diff --git a/services/core-ai/src/tools/openapi_discovery.py b/services/core-ai/src/tools/openapi_discovery.py new file mode 100644 index 0000000..b54518e --- /dev/null +++ b/services/core-ai/src/tools/openapi_discovery.py @@ -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 diff --git a/services/core-ai/src/tools/registry.py b/services/core-ai/src/tools/registry.py index 0628fbf..12c6608 100644 --- a/services/core-ai/src/tools/registry.py +++ b/services/core-ai/src/tools/registry.py @@ -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: