# app/ai/tools/dynamic_executor.py """ Dynamic Tool Executor — handles execution of 'api' and 'db_query' tools that are defined in tools_config.yaml but have no hardcoded handler in brain.py. """ from typing import Any, Dict, Tuple import httpx from structlog import get_logger from app.config import settings logger = get_logger(__name__) # Base URL for internal API calls (same server). # Uses INTERNAL_BASE_URL env var when set (e.g. in Docker/cloud), otherwise # falls back to localhost so local development works without any configuration. _INTERNAL_BASE = settings.INTERNAL_BASE_URL or f"http://127.0.0.1:{settings.SERVER_PORT}" async def execute_dynamic_tool( tool_name: str, tool_def: Dict[str, Any], params: Dict[str, Any], state: Any, # AgentState — avoid circular import ) -> Tuple[bool, str, Any]: """ Execute a dynamic (non-builtin) tool. Returns the same (success, message, result_data) tuple as execute_tool(). """ handler_type = tool_def.get("handler_type", "") try: if handler_type == "api": return await _execute_api_tool(tool_name, tool_def, params, state) elif handler_type == "db_query": return await _execute_db_query_tool(tool_name, tool_def, params, state) else: logger.warning("Unknown dynamic handler_type", tool=tool_name, handler_type=handler_type) return False, f"Unsupported handler type: {handler_type}", None except Exception as exc: logger.error("Dynamic tool execution failed", tool=tool_name, error=str(exc), exc_info=exc) return False, f"Tool {tool_name} failed: {str(exc)}", None # ── API Tool Executor ──────────────────────────────────────── async def _execute_api_tool( tool_name: str, tool_def: Dict[str, Any], params: Dict[str, Any], state: Any, ) -> Tuple[bool, str, Any]: """ Call an internal API endpoint defined in handler_config. Supports GET and POST. Path parameters are interpolated from params. """ config = tool_def.get("handler_config", {}) method = config.get("method", "GET").upper() endpoint = config.get("endpoint", "") auth_required = config.get("auth_required", False) if not endpoint: return False, f"No endpoint configured for tool {tool_name}", None # Interpolate path parameters (e.g., /api/users/public/{user_id}) url = _INTERNAL_BASE + endpoint used_path_params = set() for key, value in params.items(): placeholder = "{" + key + "}" if placeholder in url: url = url.replace(placeholder, str(value)) used_path_params.add(key) # Remaining params become query params (GET) or body (POST) remaining_params = {k: v for k, v in params.items() if k not in used_path_params} # Build headers headers = {"Content-Type": "application/json"} if auth_required: # Use the user_id to create an internal service token # or pass user context via a trusted header headers["X-Internal-User-Id"] = state.user_id headers["X-Internal-Service"] = "aida-agent" logger.info( "Executing API tool", tool=tool_name, method=method, url=url, params=remaining_params, ) async with httpx.AsyncClient(timeout=15.0) as client: if method == "GET": resp = await client.get(url, params=remaining_params, headers=headers) elif method == "POST": resp = await client.post(url, json=remaining_params, headers=headers) elif method == "PUT": resp = await client.put(url, json=remaining_params, headers=headers) elif method == "DELETE": resp = await client.delete(url, params=remaining_params, headers=headers) else: return False, f"Unsupported HTTP method: {method}", None if resp.status_code >= 400: logger.warning( "API tool returned error", tool=tool_name, status=resp.status_code, body=resp.text[:300], ) return False, f"API error ({resp.status_code}): {resp.text[:200]}", None try: data = resp.json() except Exception: data = {"raw": resp.text[:500]} return True, f"Tool {tool_name} executed successfully", data # ── DB Query Tool Executor ─────────────────────────────────── async def _execute_db_query_tool( tool_name: str, tool_def: Dict[str, Any], params: Dict[str, Any], state: Any, ) -> Tuple[bool, str, Any]: """ Execute a MongoDB aggregation pipeline defined in handler_config. Supports optional parameter injection (e.g., location filter). """ from app.database import get_db config = tool_def.get("handler_config", {}) collection_name = config.get("collection", "") pipeline_template = config.get("pipeline", []) result_format = config.get("result_format", "") if not collection_name or not pipeline_template: return False, f"Incomplete db_query config for tool {tool_name}", None # Deep-copy pipeline so we don't mutate the config import copy pipeline = copy.deepcopy(pipeline_template) # Inject optional filters from params location = params.get("location") if location: # Prepend a location filter match stage location_match = {"$match": {"location": {"$regex": location, "$options": "i"}}} pipeline.insert(0, location_match) logger.info( "Executing DB query tool", tool=tool_name, collection=collection_name, pipeline_stages=len(pipeline), ) db = await get_db() collection = db[collection_name] cursor = collection.aggregate(pipeline) results = await cursor.to_list(length=100) if not results: return True, "No data found matching the criteria", [] # Format results using the template if provided if result_format: formatted_lines = [] for doc in results: try: line = result_format.format(**doc) formatted_lines.append(line) except (KeyError, ValueError): formatted_lines.append(str(doc)) summary = "\n".join(formatted_lines) else: summary = str(results[:10]) return True, summary, results