58 lines
2.2 KiB
Python
58 lines
2.2 KiB
Python
from typing import Any
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.ai.tool_registry import ToolRegistry
|
|
from app.config import get_settings
|
|
from app.domain.schemas import ActionStatus
|
|
from app.infrastructure.database import PendingAction
|
|
|
|
|
|
class ActionExecutor:
|
|
"""Generates approved API calls without sending them to the Wed backend."""
|
|
|
|
def __init__(self, registry: ToolRegistry) -> None:
|
|
self.registry = registry
|
|
self.settings = get_settings()
|
|
self.base_url = self.settings.wed_api_base_url.rstrip("/")
|
|
|
|
async def execute(self, session: Session, action: PendingAction) -> PendingAction:
|
|
if action.status != ActionStatus.APPROVED:
|
|
raise ValueError("Only approved actions can be converted into API calls.")
|
|
tool = self.registry.get(action.tool)
|
|
if not tool:
|
|
raise ValueError(f"Unknown tool: {action.tool}")
|
|
path = tool.path
|
|
body = dict(action.payload)
|
|
for param in tool.path_params:
|
|
value = body.pop(param, None)
|
|
if value is None:
|
|
raise ValueError(f"Missing path parameter: {param}")
|
|
path = path.replace(f"{{{param}}}", str(value))
|
|
headers = {}
|
|
if self.settings.wed_api_bearer_token:
|
|
headers["Authorization"] = "Bearer <configured WED_API_BEARER_TOKEN>"
|
|
action.execution_result = {
|
|
"generated_only": True,
|
|
"method": tool.method,
|
|
"url": f"{self.base_url}{path}",
|
|
"path": path,
|
|
"headers": headers,
|
|
"json": body,
|
|
"curl": self._curl(tool.method, f"{self.base_url}{path}", headers, body),
|
|
}
|
|
session.add(action)
|
|
session.commit()
|
|
session.refresh(action)
|
|
return action
|
|
|
|
def _curl(self, method: str, url: str, headers: dict[str, str], body: dict[str, Any]) -> str:
|
|
import json
|
|
import shlex
|
|
|
|
parts = ["curl", "-X", method, url, "-H", "Content-Type: application/json"]
|
|
for key, value in headers.items():
|
|
parts.extend(["-H", f"{key}: {value}"])
|
|
parts.extend(["--data", json.dumps(body, default=str)])
|
|
return " ".join(shlex.quote(part) for part in parts)
|