app/ai/tools/dynamic_executor.py

# 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