1292 lines
60 KiB
Python
1292 lines
60 KiB
Python
# backend/src/agent/tools.py
|
|
# #region AgentChat.Tools [C:4] [TYPE Module] [SEMANTICS agent-chat,tools,langchain]
|
|
# @defgroup AgentChat Native LangChain @tool functions.
|
|
# @RATIONALE LangChain @tool decorator chosen over direct FastAPI calls for LangGraph compatibility — tools are auto-registered in the agent's tool-calling loop.
|
|
# @REJECTED StructuredTool wrapping — native @tool is the single source of truth.
|
|
# @INVARIANT Every @tool function MUST have a Python docstring (triple-quoted string immediately after the signature). LangChain raises ValueError("Function must have a docstring if description not provided.") when args_schema is provided without a description param AND the function lacks a docstring. This is a runtime blocker — the entire Gradio agent fails to import when ANY single tool violates this invariant. The GRACE @BRIEF comment is NOT a substitute — it's structural metadata invisible to the Python runtime. Rule: BEFORE decorating with @tool(args_schema=...), ensure `"""One-line description."""` is present on the next line.
|
|
|
|
import asyncio
|
|
from contextvars import ContextVar
|
|
import json
|
|
import os
|
|
from typing import Any
|
|
|
|
import httpx
|
|
from langchain_core.tools import tool
|
|
from pydantic import BaseModel, Field
|
|
|
|
from src.agent._config import FASTAPI_URL, SERVICE_JWT as _SERVICE_JWT
|
|
from src.agent.context import get_service_jwt, get_user_jwt, get_user_role
|
|
from src.core.logger import logger
|
|
|
|
TOOL_RESPONSE_LIMIT = 4000
|
|
TOOL_TIMEOUT_SECONDS = 30
|
|
_TOOL_RETRY_EVENTS: ContextVar[list[dict[str, Any]] | None] = ContextVar(
|
|
"agent_tool_retry_events",
|
|
default=None,
|
|
)
|
|
|
|
# ── Internal helpers ─────────────────────────────────────────────
|
|
|
|
# #region AgentChat.Tools.RetryEvents [C:2] [TYPE Function] [SEMANTICS agent-chat,tools,retry,sse]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Context-local retry event buffer used by Gradio stream to emit tool_retry metadata.
|
|
def start_tool_retry_event_buffer() -> None:
|
|
"""Initialise an empty retry-event buffer for the current agent request."""
|
|
_TOOL_RETRY_EVENTS.set([])
|
|
|
|
|
|
def drain_tool_retry_events() -> list[dict[str, Any]]:
|
|
"""Return and clear pending tool_retry events for the current request."""
|
|
events = _TOOL_RETRY_EVENTS.get()
|
|
if not events:
|
|
return []
|
|
drained = list(events)
|
|
events.clear()
|
|
return drained
|
|
|
|
|
|
def _queue_tool_retry_event(tool_name: str, attempt: int, max_attempts: int) -> None:
|
|
"""Queue a tool_retry event if a request-local buffer is active."""
|
|
events = _TOOL_RETRY_EVENTS.get()
|
|
if events is None:
|
|
return
|
|
events.append({
|
|
"content": f"🔁 {tool_name} retry {attempt}/{max_attempts}",
|
|
"metadata": {
|
|
"type": "tool_retry",
|
|
"tool": tool_name,
|
|
"attempt": attempt,
|
|
"max_attempts": max_attempts,
|
|
},
|
|
})
|
|
# #endregion AgentChat.Tools.RetryEvents
|
|
|
|
# #region AgentChat.Tools.DualAuthHeaders [C:2] [TYPE Function] [SEMANTICS agent-chat,tools,auth,helper]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Build dual-identity auth headers for tool→FastAPI calls per FR-007/FR-019.
|
|
def _dual_auth_headers() -> dict[str, str]:
|
|
user_jwt = get_user_jwt() or ""
|
|
svc_jwt = get_service_jwt() or os.getenv("SERVICE_JWT", "") or _SERVICE_JWT
|
|
headers = {}
|
|
if svc_jwt:
|
|
headers["Authorization"] = f"Bearer {svc_jwt}"
|
|
if user_jwt:
|
|
headers["X-User-JWT"] = user_jwt
|
|
elif user_jwt:
|
|
headers["Authorization"] = f"Bearer {user_jwt}"
|
|
return headers
|
|
# #endregion AgentChat.Tools.DualAuthHeaders
|
|
|
|
|
|
# #region AgentChat.Tools.TrimResponse [C:1] [TYPE Function] [SEMANTICS agent-chat,tools,helper]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Truncate response text to TOOL_RESPONSE_LIMIT characters.
|
|
def _trim_response(text: str, limit: int = TOOL_RESPONSE_LIMIT) -> str:
|
|
if len(text) <= limit:
|
|
return text
|
|
return f"{text[:limit]}\n... response truncated ..."
|
|
# #endregion AgentChat.Tools.TrimResponse
|
|
|
|
|
|
_SENSITIVE_FIELD_FRAGMENTS = ("password", "secret", "token", "api_key", "apikey")
|
|
|
|
|
|
# #region AgentChat.Tools.RedactSensitive [C:2] [TYPE Function] [SEMANTICS agent-chat,tools,security,helper]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Recursively redact sensitive fields (passwords, tokens, keys) from dict/list values.
|
|
def _redact_sensitive_fields(value: Any) -> Any:
|
|
if isinstance(value, list):
|
|
return [_redact_sensitive_fields(item) for item in value]
|
|
if isinstance(value, dict):
|
|
redacted: dict[str, Any] = {}
|
|
for key, item in value.items():
|
|
lowered = str(key).lower()
|
|
if any(fragment in lowered for fragment in _SENSITIVE_FIELD_FRAGMENTS):
|
|
redacted[key] = "[redacted]"
|
|
else:
|
|
redacted[key] = _redact_sensitive_fields(item)
|
|
return redacted
|
|
return value
|
|
# #endregion AgentChat.Tools.RedactSensitive
|
|
|
|
|
|
# #region AgentChat.Tools.SafeJsonText [C:2] [TYPE Function] [SEMANTICS agent-chat,tools,security,helper]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Parse JSON text and redact sensitive fields before re-serializing.
|
|
def _safe_json_text(text: str) -> str:
|
|
try:
|
|
data = json.loads(text)
|
|
except json.JSONDecodeError:
|
|
return text
|
|
return json.dumps(_redact_sensitive_fields(data), ensure_ascii=False)
|
|
# #endregion AgentChat.Tools.SafeJsonText
|
|
|
|
|
|
# #region AgentChat.Tools.PermissionGuard [C:2] [TYPE Function] [SEMANTICS agent-chat,tools,security,rbac]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Enforce invocation-time RBAC before mutating tool side effects.
|
|
def _guard_tool_permission(tool_name: str) -> None:
|
|
from src.agent._tool_filter import _TOOL_PERMISSIONS, enforce_tool_permission
|
|
|
|
user_role = get_user_role()
|
|
if enforce_tool_permission(tool_name, user_role):
|
|
return
|
|
required_role = (_TOOL_PERMISSIONS.get(tool_name) or ["admin"])[0]
|
|
raise PermissionError(
|
|
f"PERMISSION_DENIED:{tool_name}:{required_role}:{user_role}"
|
|
)
|
|
# #endregion AgentChat.Tools.PermissionGuard
|
|
|
|
|
|
# #region AgentChat.Tools.Retry [C:3] [TYPE Function] [SEMANTICS agent-chat,tools,retry,http]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Fixed-delay retry wrapper for read-only tool HTTP calls on transient errors.
|
|
# @PRE Tool is read-only (risk_level="safe"). Error is 5xx or httpx.ConnectError.
|
|
# @POST Retries once with fixed 1s asyncio.sleep. On success: returns response. On exhaust: raises with retry_exhausted=True.
|
|
# @TEST_EDGE: first_attempt_502 -> Auto-retries once, succeeds.
|
|
# @TEST_EDGE: both_attempts_502 -> Raises error with retry_exhausted=True.
|
|
# @TEST_EDGE: write_tool_502 -> No retry, raises immediately.
|
|
# @RATIONALE 1 retry with 1s delay catches ~80% of transient failures. Read-only only.
|
|
# @REJECTED Exponential backoff — single retry cannot be exponential.
|
|
# @REJECTED Retry all tools — write operations not idempotent.
|
|
|
|
async def _retry_read_tool(
|
|
tool_name: str,
|
|
tool_fn,
|
|
*args,
|
|
**kwargs,
|
|
):
|
|
"""Auto-retry read-only tool on transient HTTP errors. Fixed 1s delay, 1 retry.
|
|
|
|
Returns the tool result on success. Raises on failure.
|
|
Caller is responsible for SSE event emission.
|
|
"""
|
|
max_attempts = 2
|
|
last_error = None
|
|
for attempt in range(1, max_attempts + 1):
|
|
try:
|
|
result = await tool_fn(*args, **kwargs)
|
|
return result
|
|
except (httpx.HTTPStatusError, httpx.ConnectError, httpx.ReadTimeout) as e:
|
|
last_error = e
|
|
if attempt < max_attempts:
|
|
_queue_tool_retry_event(tool_name, attempt, max_attempts)
|
|
logger.reason("Tool retry", payload={"tool": tool_name, "attempt": attempt, "error": str(e)[:100]}, extra={"src": "AgentChat.Tools.Retry"})
|
|
await asyncio.sleep(1)
|
|
continue
|
|
# Exhausted
|
|
logger.explore("Tool retry exhausted", payload={"tool": tool_name, "attempts": max_attempts}, error=str(last_error), extra={"src": "AgentChat.Tools.Retry"})
|
|
raise last_error
|
|
# #endregion AgentChat.Tools.Retry
|
|
|
|
|
|
# #region AgentChat.Tools.Summarise [C:2] [TYPE Function] [SEMANTICS agent-chat,tools,summarise,response]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Structured truncation — top-N items + total count for large tool responses.
|
|
# @POST When response >4000 chars + JSON array: "Found N items:\n - item1\n ...\n + (N-5) more."
|
|
# When >4000 chars + JSON object: "Result keys: {keys}. Sample values: ..."
|
|
# When >4000 chars + not JSON: truncate at sentence boundary + "..."
|
|
# When ≤4000 chars: return original.
|
|
# @RATIONALE Structured truncation preserves context for LLM vs hard cut.
|
|
|
|
def _summarise_response(text: str, limit: int = 4000) -> str:
|
|
"""Structured truncation for large tool responses."""
|
|
if len(text) <= limit:
|
|
return text
|
|
|
|
try:
|
|
data = json.loads(text)
|
|
if isinstance(data, list):
|
|
total = len(data)
|
|
top = data[:5]
|
|
lines = [f"Found {total} items:"]
|
|
for item in top:
|
|
lines.append(f" - {str(item)[:120]}")
|
|
if total > 5:
|
|
lines.append(f" ... and {total - 5} more items.")
|
|
return "\n".join(lines)
|
|
elif isinstance(data, dict):
|
|
keys = list(data.keys())[:10]
|
|
sample = {k: str(data[k])[:80] for k in keys}
|
|
return f"Result keys: {keys}. Sample: {sample}"
|
|
except (json.JSONDecodeError, ValueError):
|
|
pass
|
|
|
|
# Non-JSON: truncate at last sentence boundary before limit
|
|
truncated = text[:limit]
|
|
last_period = max(truncated.rfind("."), truncated.rfind("\n"))
|
|
if last_period > limit // 2:
|
|
return truncated[:last_period + 1] + "..."
|
|
return truncated + "..."
|
|
# #endregion AgentChat.Tools.Summarise
|
|
|
|
|
|
# #region AgentChat.Tools.Timeout [C:3] [TYPE Function] [SEMANTICS agent-chat,tools,timeout]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Configurable timeout wrapper (default 30s). Write tools: retryable=false.
|
|
# @POST Returns tool result within timeout. On TimeoutError: yields tool_timeout SSE.
|
|
# @TEST_EDGE: complete_under_timeout→normal, read_exceed→retryable timeout, write_exceed→retryable=false
|
|
|
|
async def _execute_with_timeout(tool_name: str, tool_fn, is_write: bool = False, timeout_s: int = 30):
|
|
"""Execute tool with configurable timeout. Write tools: no retry on timeout.
|
|
|
|
Returns tool result on success. Raises asyncio.TimeoutError on timeout.
|
|
Caller is responsible for SSE event emission.
|
|
"""
|
|
try:
|
|
return await asyncio.wait_for(tool_fn(), timeout=timeout_s)
|
|
except TimeoutError:
|
|
logger.explore("Tool timeout", payload={"tool": tool_name, "timeout_s": timeout_s, "is_write": is_write}, extra={"src": "AgentChat.Tools.Timeout"})
|
|
raise
|
|
# #endregion AgentChat.Tools.Timeout
|
|
|
|
|
|
# #region AgentChat.Tools.HttpGet [C:2] [TYPE Function] [SEMANTICS agent-chat,tools,http,helper]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Async HTTP GET to FastAPI with dual-auth headers.
|
|
async def _get(path: str, params: dict[str, Any] | None = None) -> httpx.Response:
|
|
async def _request() -> httpx.Response:
|
|
async with httpx.AsyncClient(timeout=TOOL_TIMEOUT_SECONDS) as client:
|
|
resp = await client.get(
|
|
f"{FASTAPI_URL}{path}",
|
|
params=params,
|
|
headers=_dual_auth_headers(),
|
|
)
|
|
if resp.status_code in {502, 503, 504}:
|
|
raise httpx.HTTPStatusError(
|
|
f"Transient FastAPI error {resp.status_code}",
|
|
request=resp.request,
|
|
response=resp,
|
|
)
|
|
return resp
|
|
|
|
try:
|
|
return await _execute_with_timeout(path, lambda: _retry_read_tool(path, _request), timeout_s=TOOL_TIMEOUT_SECONDS)
|
|
except TimeoutError as exc:
|
|
raise TimeoutError(f"Read tool timed out after {TOOL_TIMEOUT_SECONDS}s: {path}") from exc
|
|
# #endregion AgentChat.Tools.HttpGet
|
|
|
|
|
|
# #region AgentChat.Tools.HttpPost [C:2] [TYPE Function] [SEMANTICS agent-chat,tools,http,helper]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Async HTTP POST to FastAPI with dual-auth headers.
|
|
async def _post(
|
|
path: str,
|
|
payload: dict[str, Any] | None = None,
|
|
params: dict[str, Any] | None = None,
|
|
) -> httpx.Response:
|
|
async def _request() -> httpx.Response:
|
|
async with httpx.AsyncClient(timeout=TOOL_TIMEOUT_SECONDS) as client:
|
|
return await client.post(
|
|
f"{FASTAPI_URL}{path}",
|
|
json=payload or {},
|
|
params=params,
|
|
headers=_dual_auth_headers(),
|
|
)
|
|
|
|
try:
|
|
return await _execute_with_timeout(path, _request, is_write=True, timeout_s=TOOL_TIMEOUT_SECONDS)
|
|
except TimeoutError as exc:
|
|
raise TimeoutError(
|
|
f"Write tool timed out after {TOOL_TIMEOUT_SECONDS}s; operation status unknown, check status: {path}"
|
|
) from exc
|
|
# #endregion AgentChat.Tools.HttpPost
|
|
|
|
|
|
# #region AgentChat.Tools.ApiResult [C:2] [TYPE Function] [SEMANTICS agent-chat,tools,http,helper]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Extract result text from HTTP response, trimming to limit.
|
|
def _api_result(resp: httpx.Response, ok_statuses: set[int] | None = None) -> str:
|
|
ok_statuses = ok_statuses or {200, 201, 202}
|
|
if resp.status_code not in ok_statuses:
|
|
return f"Error {resp.status_code}: {resp.text}"
|
|
return _summarise_response(resp.text, TOOL_RESPONSE_LIMIT)
|
|
# #endregion AgentChat.Tools.ApiResult
|
|
|
|
|
|
# #region AgentChat.Tools.ResolveProvider [C:2] [TYPE Function] [SEMANTICS agent-chat,tools,llm,helper]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Resolve active LLM provider ID — explicit or first active from /api/llm/providers.
|
|
async def _resolve_active_provider_id(provider_id: str | None = None) -> str | None:
|
|
if provider_id:
|
|
return provider_id
|
|
resp = await _get("/api/llm/providers")
|
|
if resp.status_code != 200:
|
|
return None
|
|
try:
|
|
providers = resp.json()
|
|
except json.JSONDecodeError:
|
|
return None
|
|
active = next((p for p in providers if p.get("is_active")), None)
|
|
if active:
|
|
return str(active.get("id"))
|
|
if providers:
|
|
return str(providers[0].get("id"))
|
|
return None
|
|
# #endregion AgentChat.Tools.ResolveProvider
|
|
|
|
|
|
# #region AgentChat.Tools.DashboardIds [C:1] [TYPE Function] [SEMANTICS agent-chat,tools,helper]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Merge single dashboard_id + list of dashboard_ids into deduplicated int list.
|
|
def _dashboard_ids(dashboard_id: int | str | None, dashboard_ids: list[int] | list[str] | None) -> list[int]:
|
|
values: list[int] = []
|
|
for raw in dashboard_ids or []:
|
|
values.append(int(raw))
|
|
if dashboard_id is not None and str(dashboard_id).strip():
|
|
values.append(int(dashboard_id))
|
|
return list(dict.fromkeys(values))
|
|
# #endregion AgentChat.Tools.DashboardIds
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# Tool definitions — each wrapped in a #region contract
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
# #region AgentChat.Tools.SearchDashboards.Schema [C:1] [TYPE Class] [SEMANTICS agent-chat,tools,schema,dashboard]
|
|
class SearchDashboardsInput(BaseModel):
|
|
query: str = Field(description="Search query for dashboard name")
|
|
env_id: str | None = Field(default=None, description="Environment ID (e.g. 'prod', 'ss-dev')")
|
|
# #endregion AgentChat.Tools.SearchDashboards.Schema
|
|
|
|
|
|
# #region AgentChat.Tools.SearchDashboards [C:3] [TYPE Function] [SEMANTICS agent-chat,tools,search,dashboard]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Search and list dashboards by name, with optional environment filter.
|
|
# @PRE User authenticated via dual-identity JWT.
|
|
# @POST Returns formatted dashboard list string.
|
|
# @SIDE_EFFECT HTTP GET to FastAPI /api/dashboards.
|
|
@tool(args_schema=SearchDashboardsInput)
|
|
async def search_dashboards(query: str, env_id: str | None = None) -> str:
|
|
"""Search and list dashboards by name, with optional environment filter."""
|
|
logger.reason("Search dashboards", payload={"query": query, "env_id": env_id},
|
|
extra={"src": "AgentChat.Tools.SearchDashboards"})
|
|
params = {"q": query, "env_id": env_id or ""}
|
|
resp = await _get("/api/dashboards", params=params)
|
|
if resp.status_code != 200:
|
|
logger.explore("Dashboard search failed",
|
|
payload={"status": resp.status_code, "query": query},
|
|
error=resp.text[:200],
|
|
extra={"src": "AgentChat.Tools.SearchDashboards"})
|
|
return f"Error {resp.status_code}: Unable to fetch dashboards. {resp.text}"
|
|
|
|
try:
|
|
data = resp.json()
|
|
dashboards = data.get("dashboards", [])
|
|
total = data.get("total", 0)
|
|
|
|
if total == 0:
|
|
logger.reflect("No dashboards found",
|
|
payload={"query": query, "total": 0},
|
|
extra={"src": "AgentChat.Tools.SearchDashboards"})
|
|
return f"No dashboards found matching '{query}' in environment '{env_id or 'default'}'."
|
|
|
|
lines = [f"Found {total} dashboard(s) in environment '{env_id or 'default'}':"]
|
|
for db in dashboards:
|
|
title = db.get("title", "Untitled")
|
|
owners = ", ".join(db.get("owners", [])) or "N/A"
|
|
modified = db.get("last_modified", "")[:10] if db.get("last_modified") else "unknown"
|
|
lines.append(f" - {title}")
|
|
lines.append(f" Owners: {owners}")
|
|
lines.append(f" Last modified: {modified}")
|
|
logger.reflect("Dashboards listed",
|
|
payload={"query": query, "total": total},
|
|
extra={"src": "AgentChat.Tools.SearchDashboards"})
|
|
return "\n".join(lines)
|
|
except (json.JSONDecodeError, KeyError) as e:
|
|
logger.explore("Failed to parse dashboards response",
|
|
payload={"query": query}, error=str(e),
|
|
extra={"src": "AgentChat.Tools.SearchDashboards"})
|
|
return f"Could not parse dashboards response: {e}"
|
|
# #endregion AgentChat.Tools.SearchDashboards
|
|
|
|
|
|
# #region AgentChat.Tools.HealthSummary.Schema [C:1] [TYPE Class] [SEMANTICS agent-chat,tools,schema,health]
|
|
class HealthSummaryInput(BaseModel):
|
|
env_id: str | None = Field(default=None, description="Environment ID (e.g. 'prod', 'ss-dev')")
|
|
# #endregion AgentChat.Tools.HealthSummary.Schema
|
|
|
|
|
|
# #region AgentChat.Tools.HealthSummary [C:3] [TYPE Function] [SEMANTICS agent-chat,tools,health,summary]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Get system health summary — dashboard validation status, recent failures.
|
|
# @PRE User authenticated via dual-identity JWT.
|
|
# @POST Returns health summary text (trimmed to 2000 chars).
|
|
# @SIDE_EFFECT HTTP GET to FastAPI /api/health/summary.
|
|
@tool(args_schema=HealthSummaryInput)
|
|
async def get_health_summary(env_id: str | None = None) -> str:
|
|
"""Get system health summary — dashboard validation status, recent failures."""
|
|
logger.reason("Get health summary", payload={"env_id": env_id},
|
|
extra={"src": "AgentChat.Tools.HealthSummary"})
|
|
resp = await _get("/api/health/summary", params={"environment_id": env_id or ""})
|
|
if resp.status_code != 200:
|
|
logger.explore("Health check failed",
|
|
payload={"status": resp.status_code},
|
|
error=resp.text[:200],
|
|
extra={"src": "AgentChat.Tools.HealthSummary"})
|
|
return f"Health check failed: HTTP {resp.status_code}"
|
|
logger.reflect("Health summary retrieved",
|
|
payload={"status": resp.status_code},
|
|
extra={"src": "AgentChat.Tools.HealthSummary"})
|
|
return _trim_response(resp.text, 2000)
|
|
# #endregion AgentChat.Tools.HealthSummary
|
|
|
|
|
|
# #region AgentChat.Tools.ListEnvironments [C:3] [TYPE Function] [SEMANTICS agent-chat,tools,environment,list]
|
|
# @ingroup AgentChat
|
|
# @BRIEF List configured deployment environments.
|
|
# @PRE User authenticated via dual-identity JWT.
|
|
# @POST Returns JSON string of environments (sensitive fields redacted).
|
|
# @SIDE_EFFECT HTTP GET to FastAPI /api/settings/environments.
|
|
@tool
|
|
async def list_environments() -> str:
|
|
"""List configured deployment environments."""
|
|
logger.reason("List environments", extra={"src": "AgentChat.Tools.ListEnvironments"})
|
|
resp = await _get("/api/settings/environments")
|
|
result = _api_result(resp)
|
|
logger.reflect("Environments listed", payload={"status": resp.status_code},
|
|
extra={"src": "AgentChat.Tools.ListEnvironments"})
|
|
return _trim_response(_safe_json_text(result))
|
|
# #endregion AgentChat.Tools.ListEnvironments
|
|
|
|
|
|
# #region AgentChat.Tools.TaskStatus [C:3] [TYPE Function] [SEMANTICS agent-chat,tools,task,status]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Check the status of a background task by its task_id.
|
|
# @PRE User authenticated via dual-identity JWT.
|
|
# @POST Returns task status text from FastAPI.
|
|
# @SIDE_EFFECT HTTP GET to FastAPI /api/tasks/{task_id}.
|
|
@tool
|
|
async def get_task_status(task_id: str) -> str:
|
|
"""Check the status of a background task by its task_id."""
|
|
logger.reason("Get task status", payload={"task_id": task_id},
|
|
extra={"src": "AgentChat.Tools.TaskStatus"})
|
|
resp = await _get(f"/api/tasks/{task_id}")
|
|
result = _api_result(resp)
|
|
logger.reflect("Task status retrieved", payload={"task_id": task_id, "status": resp.status_code},
|
|
extra={"src": "AgentChat.Tools.TaskStatus"})
|
|
return result
|
|
# #endregion AgentChat.Tools.TaskStatus
|
|
|
|
|
|
# #region AgentChat.Tools.ShowCapabilities [C:2] [TYPE Function] [SEMANTICS agent-chat,tools,capabilities,introspection]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Show all tools available to the agent chat — names + descriptions.
|
|
@tool
|
|
async def show_capabilities() -> str:
|
|
"""Show all tools available to the agent chat — names + descriptions."""
|
|
lines = ["Available LangChain tools:"]
|
|
for tool_obj in get_all_tools():
|
|
lines.append(f"- {tool_obj.name}: {tool_obj.description or ''}")
|
|
return "\n".join(lines)
|
|
# #endregion AgentChat.Tools.ShowCapabilities
|
|
|
|
|
|
# #region AgentChat.Tools.ListLlmProviders [C:3] [TYPE Function] [SEMANTICS agent-chat,tools,llm,providers]
|
|
# @ingroup AgentChat
|
|
# @BRIEF List configured LLM providers.
|
|
# @PRE User authenticated via dual-identity JWT.
|
|
# @POST Returns provider list text.
|
|
# @SIDE_EFFECT HTTP GET to FastAPI /api/llm/providers.
|
|
@tool
|
|
async def list_llm_providers() -> str:
|
|
"""List configured LLM providers."""
|
|
logger.reason("List LLM providers", extra={"src": "AgentChat.Tools.ListLlmProviders"})
|
|
resp = await _get("/api/llm/providers")
|
|
result = _api_result(resp)
|
|
logger.reflect("LLM providers listed", payload={"status": resp.status_code},
|
|
extra={"src": "AgentChat.Tools.ListLlmProviders"})
|
|
return result
|
|
# #endregion AgentChat.Tools.ListLlmProviders
|
|
|
|
|
|
# #region AgentChat.Tools.LlmStatus [C:3] [TYPE Function] [SEMANTICS agent-chat,tools,llm,status]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Check whether the LLM runtime is configured and usable.
|
|
# @PRE User authenticated via dual-identity JWT.
|
|
# @POST Returns LLM status text.
|
|
# @SIDE_EFFECT HTTP GET to FastAPI /api/llm/status.
|
|
@tool
|
|
async def get_llm_status() -> str:
|
|
"""Check whether the LLM runtime is configured and usable."""
|
|
logger.reason("Get LLM status", extra={"src": "AgentChat.Tools.LlmStatus"})
|
|
resp = await _get("/api/llm/status")
|
|
result = _api_result(resp)
|
|
logger.reflect("LLM status retrieved", payload={"status": resp.status_code},
|
|
extra={"src": "AgentChat.Tools.LlmStatus"})
|
|
return result
|
|
# #endregion AgentChat.Tools.LlmStatus
|
|
|
|
|
|
# #region AgentChat.Tools.RunBackup.Schema [C:1] [TYPE Class] [SEMANTICS agent-chat,tools,schema,backup]
|
|
class RunBackupInput(BaseModel):
|
|
environment_id: str = Field(description="Environment ID for the backup")
|
|
dashboard_id: int | None = Field(default=None, description="Optional single dashboard ID")
|
|
dashboard_ids: list[int] | None = Field(default=None, description="Optional list of dashboard IDs")
|
|
# #endregion AgentChat.Tools.RunBackup.Schema
|
|
|
|
|
|
# #region AgentChat.Tools.RunBackup [C:3] [TYPE Function] [SEMANTICS agent-chat,tools,backup,operation]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Run a Superset backup for an environment, optionally scoped to dashboard IDs.
|
|
# @PRE User authenticated via dual-identity JWT.
|
|
# @POST Returns task creation result (201).
|
|
# @SIDE_EFFECT HTTP POST to FastAPI /api/tasks (superset-backup plugin).
|
|
@tool(args_schema=RunBackupInput)
|
|
async def run_backup(
|
|
environment_id: str,
|
|
dashboard_id: int | None = None,
|
|
dashboard_ids: list[int] | None = None,
|
|
) -> str:
|
|
"""Run a Superset backup for an environment, optionally scoped to dashboard IDs."""
|
|
_guard_tool_permission("run_backup")
|
|
vars = {"env_id": environment_id, "dashboard_id": dashboard_id, "dashboard_ids": dashboard_ids}
|
|
logger.reason("Run backup", payload=vars, extra={"src": "AgentChat.Tools.RunBackup"})
|
|
params: dict[str, Any] = {"environment_id": environment_id}
|
|
ids_l = _dashboard_ids(dashboard_id, dashboard_ids)
|
|
if ids_l:
|
|
params["dashboard_ids"] = ids_l
|
|
resp = await _post("/api/tasks", {"plugin_id": "superset-backup", "params": params})
|
|
result = _api_result(resp, {201})
|
|
logger.reflect("Backup task created", payload={"status": resp.status_code},
|
|
extra={"src": "AgentChat.Tools.RunBackup"})
|
|
return result
|
|
# #endregion AgentChat.Tools.RunBackup
|
|
|
|
|
|
# #region AgentChat.Tools.ExecuteMigration.Schema [C:1] [TYPE Class] [SEMANTICS agent-chat,tools,schema,migration]
|
|
class ExecuteMigrationInput(BaseModel):
|
|
source_env_id: str = Field(description="Source environment ID")
|
|
target_env_id: str = Field(description="Target environment ID")
|
|
selected_ids: list[int] = Field(description="Dashboard IDs to migrate")
|
|
replace_db_config: bool = Field(default=False, description="Replace DB configuration during migration")
|
|
fix_cross_filters: bool = Field(default=True, description="Fix cross filters during migration")
|
|
# #endregion AgentChat.Tools.ExecuteMigration.Schema
|
|
|
|
|
|
# #region AgentChat.Tools.ExecuteMigration [C:3] [TYPE Function] [SEMANTICS agent-chat,tools,migration,operation]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Execute dashboard migration between two environments.
|
|
# @PRE User authenticated via dual-identity JWT.
|
|
# @POST Returns migration result text.
|
|
# @SIDE_EFFECT HTTP POST to FastAPI /api/migration/execute.
|
|
@tool(args_schema=ExecuteMigrationInput)
|
|
async def execute_migration(
|
|
source_env_id: str,
|
|
target_env_id: str,
|
|
selected_ids: list[int],
|
|
replace_db_config: bool = False,
|
|
fix_cross_filters: bool = True,
|
|
) -> str:
|
|
"""Execute dashboard migration between two environments."""
|
|
_guard_tool_permission("execute_migration")
|
|
logger.reason("Execute migration",
|
|
payload={"source": source_env_id, "target": target_env_id, "ids_count": len(selected_ids)},
|
|
extra={"src": "AgentChat.Tools.ExecuteMigration"})
|
|
payload = {
|
|
"source_env_id": source_env_id,
|
|
"target_env_id": target_env_id,
|
|
"selected_ids": selected_ids,
|
|
"replace_db_config": replace_db_config,
|
|
"fix_cross_filters": fix_cross_filters,
|
|
}
|
|
resp = await _post("/api/migration/execute", payload)
|
|
result = _api_result(resp)
|
|
logger.reflect("Migration executed", payload={"status": resp.status_code},
|
|
extra={"src": "AgentChat.Tools.ExecuteMigration"})
|
|
return result
|
|
# #endregion AgentChat.Tools.ExecuteMigration
|
|
|
|
|
|
# ── Git tools: shared input schema ──
|
|
|
|
# #region AgentChat.Tools.GitDashboardInput [C:1] [TYPE Class] [SEMANTICS agent-chat,tools,schema,git]
|
|
class GitDashboardInput(BaseModel):
|
|
dashboard_ref: str = Field(description="Dashboard ID, slug, or title resolvable by the Git API")
|
|
env_id: str | None = Field(default=None, description="Optional environment ID used to resolve dashboard_ref")
|
|
# #endregion AgentChat.Tools.GitDashboardInput
|
|
|
|
|
|
# #region AgentChat.Tools.CreateBranchInput [C:1] [TYPE Class] [SEMANTICS agent-chat,tools,schema,git,branch]
|
|
class CreateBranchInput(GitDashboardInput):
|
|
branch_name: str = Field(description="New branch name")
|
|
from_branch: str = Field(default="dev", description="Source branch")
|
|
# #endregion AgentChat.Tools.CreateBranchInput
|
|
|
|
|
|
# #region AgentChat.Tools.CreateBranch [C:3] [TYPE Function] [SEMANTICS agent-chat,tools,git,branch]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Create a branch in a dashboard Git repository.
|
|
# @PRE User authenticated via dual-identity JWT.
|
|
# @POST Returns branch creation result.
|
|
# @SIDE_EFFECT HTTP POST to FastAPI /api/git/repositories/{ref}/branches.
|
|
@tool(args_schema=CreateBranchInput)
|
|
async def create_branch(
|
|
dashboard_ref: str,
|
|
branch_name: str,
|
|
from_branch: str = "dev",
|
|
env_id: str | None = None,
|
|
) -> str:
|
|
"""Create a branch in a dashboard Git repository."""
|
|
_guard_tool_permission("create_branch")
|
|
logger.reason("Create branch",
|
|
payload={"dashboard_ref": dashboard_ref, "branch_name": branch_name, "from_branch": from_branch},
|
|
extra={"src": "AgentChat.Tools.CreateBranch"})
|
|
resp = await _post(
|
|
f"/api/git/repositories/{dashboard_ref}/branches",
|
|
{"name": branch_name, "from_branch": from_branch},
|
|
params={"env_id": env_id} if env_id else None,
|
|
)
|
|
result = _api_result(resp)
|
|
logger.reflect("Branch created", payload={"status": resp.status_code},
|
|
extra={"src": "AgentChat.Tools.CreateBranch"})
|
|
return result
|
|
# #endregion AgentChat.Tools.CreateBranch
|
|
|
|
|
|
# #region AgentChat.Tools.CommitChanges.Schema [C:1] [TYPE Class] [SEMANTICS agent-chat,tools,schema,git,commit]
|
|
class CommitChangesInput(GitDashboardInput):
|
|
message: str = Field(description="Commit message")
|
|
files: list[str] = Field(default_factory=list, description="Files to stage and commit; empty means backend default")
|
|
# #endregion AgentChat.Tools.CommitChanges.Schema
|
|
|
|
|
|
# #region AgentChat.Tools.CommitChanges [C:3] [TYPE Function] [SEMANTICS agent-chat,tools,git,commit]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Stage and commit changes in a dashboard Git repository.
|
|
# @PRE User authenticated via dual-identity JWT.
|
|
# @POST Returns commit result.
|
|
# @SIDE_EFFECT HTTP POST to FastAPI /api/git/repositories/{ref}/commit.
|
|
@tool(args_schema=CommitChangesInput)
|
|
async def commit_changes(
|
|
dashboard_ref: str,
|
|
message: str,
|
|
files: list[str] | None = None,
|
|
env_id: str | None = None,
|
|
) -> str:
|
|
"""Stage and commit changes in a dashboard Git repository."""
|
|
_guard_tool_permission("commit_changes")
|
|
logger.reason("Commit changes",
|
|
payload={"dashboard_ref": dashboard_ref, "message": message[:80]},
|
|
extra={"src": "AgentChat.Tools.CommitChanges"})
|
|
resp = await _post(
|
|
f"/api/git/repositories/{dashboard_ref}/commit",
|
|
{"message": message, "files": files or []},
|
|
params={"env_id": env_id} if env_id else None,
|
|
)
|
|
result = _api_result(resp)
|
|
logger.reflect("Changes committed", payload={"status": resp.status_code},
|
|
extra={"src": "AgentChat.Tools.CommitChanges"})
|
|
return result
|
|
# #endregion AgentChat.Tools.CommitChanges
|
|
|
|
|
|
# #region AgentChat.Tools.DeployDashboard.Schema [C:1] [TYPE Class] [SEMANTICS agent-chat,tools,schema,git,deploy]
|
|
class DeployDashboardInput(GitDashboardInput):
|
|
environment_id: str = Field(description="Target deployment environment ID")
|
|
# #endregion AgentChat.Tools.DeployDashboard.Schema
|
|
|
|
|
|
# #region AgentChat.Tools.DeployDashboard [C:3] [TYPE Function] [SEMANTICS agent-chat,tools,git,deploy]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Deploy a dashboard from Git to a target environment.
|
|
# @PRE User authenticated via dual-identity JWT.
|
|
# @POST Returns deployment result.
|
|
# @SIDE_EFFECT HTTP POST to FastAPI /api/git/repositories/{ref}/deploy.
|
|
@tool(args_schema=DeployDashboardInput)
|
|
async def deploy_dashboard(
|
|
dashboard_ref: str,
|
|
environment_id: str,
|
|
env_id: str | None = None,
|
|
) -> str:
|
|
"""Deploy a dashboard from Git to a target environment."""
|
|
_guard_tool_permission("deploy_dashboard")
|
|
logger.reason("Deploy dashboard",
|
|
payload={"dashboard_ref": dashboard_ref, "target_env": environment_id},
|
|
extra={"src": "AgentChat.Tools.DeployDashboard"})
|
|
resp = await _post(
|
|
f"/api/git/repositories/{dashboard_ref}/deploy",
|
|
{"environment_id": environment_id},
|
|
params={"env_id": env_id} if env_id else None,
|
|
)
|
|
result = _api_result(resp)
|
|
logger.reflect("Dashboard deployed", payload={"status": resp.status_code},
|
|
extra={"src": "AgentChat.Tools.DeployDashboard"})
|
|
return result
|
|
# #endregion AgentChat.Tools.DeployDashboard
|
|
|
|
|
|
# #region AgentChat.Tools.LlmDocumentation.Schema [C:1] [TYPE Class] [SEMANTICS agent-chat,tools,schema,llm,documentation]
|
|
class RunLlmDocumentationInput(BaseModel):
|
|
dataset_id: str = Field(description="Dataset ID to document")
|
|
environment_id: str = Field(description="Environment ID")
|
|
provider_id: str | None = Field(default=None, description="Optional LLM provider ID; active provider is used if omitted")
|
|
# #endregion AgentChat.Tools.LlmDocumentation.Schema
|
|
|
|
|
|
# #region AgentChat.Tools.LlmDocumentation [C:3] [TYPE Function] [SEMANTICS agent-chat,tools,llm,documentation]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Generate dataset documentation via the LLM documentation task.
|
|
# @PRE User authenticated via dual-identity JWT. LLM provider configured.
|
|
# @POST Returns task creation result (201).
|
|
# @SIDE_EFFECT HTTP POST to FastAPI /api/tasks (llm_documentation plugin).
|
|
@tool(args_schema=RunLlmDocumentationInput)
|
|
async def run_llm_documentation(
|
|
dataset_id: str,
|
|
environment_id: str,
|
|
provider_id: str | None = None,
|
|
) -> str:
|
|
"""Generate dataset documentation via the LLM documentation task."""
|
|
logger.reason("Run LLM documentation",
|
|
payload={"dataset_id": dataset_id, "environment_id": environment_id},
|
|
extra={"src": "AgentChat.Tools.LlmDocumentation"})
|
|
resolved_provider = await _resolve_active_provider_id(provider_id)
|
|
params = {
|
|
"dataset_id": str(dataset_id),
|
|
"environment_id": environment_id,
|
|
}
|
|
if resolved_provider:
|
|
params["provider_id"] = resolved_provider
|
|
resp = await _post("/api/tasks", {"plugin_id": "llm_documentation", "params": params})
|
|
result = _api_result(resp, {201})
|
|
logger.reflect("LLM documentation task created",
|
|
payload={"status": resp.status_code, "provider_id": resolved_provider},
|
|
extra={"src": "AgentChat.Tools.LlmDocumentation"})
|
|
return result
|
|
# #endregion AgentChat.Tools.LlmDocumentation
|
|
|
|
|
|
# #region AgentChat.Tools.LlmValidation.Schema [C:1] [TYPE Class] [SEMANTICS agent-chat,tools,schema,llm,validation]
|
|
class RunLlmValidationInput(BaseModel):
|
|
environment_id: str = Field(description="Environment ID")
|
|
dashboard_id: int | None = Field(default=None, description="Single dashboard ID")
|
|
dashboard_ids: list[int] | None = Field(default=None, description="Dashboard IDs to validate")
|
|
dashboard_url: str | None = Field(default=None, description="Optional full Superset dashboard URL")
|
|
provider_id: str | None = Field(default=None, description="Optional LLM provider ID; active provider is used if omitted")
|
|
name: str | None = Field(default=None, description="Optional validation policy name")
|
|
policy_id: str | None = Field(default=None, description="Existing validation policy ID to run immediately")
|
|
# #endregion AgentChat.Tools.LlmValidation.Schema
|
|
|
|
|
|
# #region AgentChat.Tools.LlmValidation [C:3] [TYPE Function] [SEMANTICS agent-chat,tools,llm,validation]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Create and immediately run an LLM dashboard validation policy, or run an existing policy.
|
|
# @PRE User authenticated via dual-identity JWT. LLM provider configured.
|
|
# @POST Returns validation task result.
|
|
# @SIDE_EFFECT HTTP POST to FastAPI /api/validation-tasks + /api/validation-tasks/{id}/run.
|
|
@tool(args_schema=RunLlmValidationInput)
|
|
async def run_llm_validation(
|
|
environment_id: str,
|
|
dashboard_id: int | None = None,
|
|
dashboard_ids: list[int] | None = None,
|
|
dashboard_url: str | None = None,
|
|
provider_id: str | None = None,
|
|
name: str | None = None,
|
|
policy_id: str | None = None,
|
|
) -> str:
|
|
"""Create and immediately run an LLM dashboard validation policy, or run an existing policy."""
|
|
if policy_id:
|
|
logger.reason("Run existing validation policy",
|
|
payload={"policy_id": policy_id},
|
|
extra={"src": "AgentChat.Tools.LlmValidation"})
|
|
resp = await _post(f"/api/validation-tasks/{policy_id}/run")
|
|
result = _api_result(resp)
|
|
logger.reflect("Validation policy run", payload={"policy_id": policy_id, "status": resp.status_code},
|
|
extra={"src": "AgentChat.Tools.LlmValidation"})
|
|
return result
|
|
|
|
ids = _dashboard_ids(dashboard_id, dashboard_ids)
|
|
logger.reason("Create LLM validation policy",
|
|
payload={"environment_id": environment_id, "dashboard_count": len(ids)},
|
|
extra={"src": "AgentChat.Tools.LlmValidation"})
|
|
sources = []
|
|
if dashboard_url:
|
|
sources.append({"type": "dashboard_url", "value": dashboard_url})
|
|
if not ids and not sources:
|
|
return "Error: provide dashboard_id, dashboard_ids, dashboard_url, or policy_id."
|
|
|
|
resolved_provider = await _resolve_active_provider_id(provider_id)
|
|
if not resolved_provider:
|
|
return "Error: no LLM provider configured. Configure or pass provider_id."
|
|
|
|
payload = {
|
|
"name": name or f"Agent LLM validation - {environment_id}",
|
|
"environment_id": environment_id,
|
|
"dashboard_ids": [str(i) for i in ids],
|
|
"provider_id": resolved_provider,
|
|
"is_active": True,
|
|
"screenshot_enabled": True,
|
|
"logs_enabled": True,
|
|
"execute_chart_data": False,
|
|
}
|
|
if sources:
|
|
payload["sources"] = sources
|
|
|
|
create_resp = await _post("/api/validation-tasks", payload)
|
|
if create_resp.status_code != 201:
|
|
return _api_result(create_resp, {201})
|
|
try:
|
|
created = create_resp.json()
|
|
created_id = created["id"]
|
|
except (json.JSONDecodeError, KeyError) as exc:
|
|
return f"Validation policy created but response could not be parsed: {exc}. Raw: {create_resp.text}"
|
|
run_resp = await _post(f"/api/validation-tasks/{created_id}/run")
|
|
result = _api_result(run_resp)
|
|
logger.reflect("Validation policy created and run",
|
|
payload={"policy_id": created_id, "status": run_resp.status_code},
|
|
extra={"src": "AgentChat.Tools.LlmValidation"})
|
|
return result
|
|
# #endregion AgentChat.Tools.LlmValidation
|
|
|
|
|
|
# #region AgentChat.Tools.ListMaintenance [C:3] [TYPE Function] [SEMANTICS agent-chat,tools,maintenance,list]
|
|
# @ingroup AgentChat
|
|
# @BRIEF List active and completed maintenance events.
|
|
# @PRE User authenticated via dual-identity JWT.
|
|
# @POST Returns maintenance events text.
|
|
# @SIDE_EFFECT HTTP GET to FastAPI /api/maintenance/events.
|
|
@tool
|
|
async def list_maintenance_events() -> str:
|
|
"""List active and completed maintenance events."""
|
|
logger.reason("List maintenance events", extra={"src": "AgentChat.Tools.ListMaintenance"})
|
|
resp = await _get("/api/maintenance/events")
|
|
result = _api_result(resp)
|
|
logger.reflect("Maintenance events listed", payload={"status": resp.status_code},
|
|
extra={"src": "AgentChat.Tools.ListMaintenance"})
|
|
return result
|
|
# #endregion AgentChat.Tools.ListMaintenance
|
|
|
|
|
|
# #region AgentChat.Tools.StartMaintenance.Schema [C:1] [TYPE Class] [SEMANTICS agent-chat,tools,schema,maintenance]
|
|
class StartMaintenanceInput(BaseModel):
|
|
tables: list[str] = Field(description="Affected table names")
|
|
environment_id: str = Field(description="Target environment ID")
|
|
start_time: str = Field(description="ISO datetime for maintenance start")
|
|
end_time: str | None = Field(default=None, description="Optional ISO datetime for maintenance end")
|
|
message: str | None = Field(default=None, description="Optional banner message")
|
|
# #endregion AgentChat.Tools.StartMaintenance.Schema
|
|
|
|
|
|
# #region AgentChat.Tools.StartMaintenance [C:3] [TYPE Function] [SEMANTICS agent-chat,tools,maintenance,start]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Start a maintenance event and apply banners to affected dashboards.
|
|
# @PRE User authenticated via dual-identity JWT.
|
|
# @POST Returns maintenance start result (200/202).
|
|
# @SIDE_EFFECT HTTP POST to FastAPI /api/maintenance/start.
|
|
@tool(args_schema=StartMaintenanceInput)
|
|
async def start_maintenance(
|
|
tables: list[str],
|
|
environment_id: str,
|
|
start_time: str,
|
|
end_time: str | None = None,
|
|
message: str | None = None,
|
|
) -> str:
|
|
"""Start a maintenance event and apply banners to affected dashboards."""
|
|
_guard_tool_permission("start_maintenance")
|
|
logger.reason("Start maintenance",
|
|
payload={"environment_id": environment_id, "tables": tables},
|
|
extra={"src": "AgentChat.Tools.StartMaintenance"})
|
|
payload = {
|
|
"tables": tables,
|
|
"environment_id": environment_id,
|
|
"start_time": start_time,
|
|
"end_time": end_time,
|
|
"message": message,
|
|
}
|
|
resp = await _post("/api/maintenance/start", payload)
|
|
result = _api_result(resp, {200, 202})
|
|
logger.reflect("Maintenance started", payload={"status": resp.status_code},
|
|
extra={"src": "AgentChat.Tools.StartMaintenance"})
|
|
return result
|
|
# #endregion AgentChat.Tools.StartMaintenance
|
|
|
|
|
|
# #region AgentChat.Tools.EndMaintenance.Schema [C:1] [TYPE Class] [SEMANTICS agent-chat,tools,schema,maintenance]
|
|
class EndMaintenanceInput(BaseModel):
|
|
maintenance_id: str | None = Field(default=None, description="Maintenance event ID; omit when end_all=True")
|
|
environment_id: str | None = Field(default=None, description="Optional environment scope for end_all")
|
|
end_all: bool = Field(default=False, description="End all active maintenance events")
|
|
# #endregion AgentChat.Tools.EndMaintenance.Schema
|
|
|
|
|
|
# #region AgentChat.Tools.EndMaintenance [C:3] [TYPE Function] [SEMANTICS agent-chat,tools,maintenance,end]
|
|
# @ingroup AgentChat
|
|
# @BRIEF End one maintenance event, or end all active events when end_all is true.
|
|
# @PRE User authenticated via dual-identity JWT.
|
|
# @POST Returns maintenance end result (202).
|
|
# @SIDE_EFFECT HTTP POST to FastAPI /api/maintenance/{id}/end or /api/maintenance/end-all.
|
|
@tool(args_schema=EndMaintenanceInput)
|
|
async def end_maintenance(
|
|
maintenance_id: str | None = None,
|
|
environment_id: str | None = None,
|
|
end_all: bool = False,
|
|
) -> str:
|
|
"""End one maintenance event, or end all active events when end_all is true."""
|
|
_guard_tool_permission("end_maintenance")
|
|
if end_all:
|
|
logger.reason("End all maintenance events",
|
|
payload={"environment_id": environment_id},
|
|
extra={"src": "AgentChat.Tools.EndMaintenance"})
|
|
payload = {"environment_id": environment_id} if environment_id else {}
|
|
resp = await _post("/api/maintenance/end-all", payload)
|
|
result = _api_result(resp, {202})
|
|
logger.reflect("All maintenance ended", payload={"status": resp.status_code},
|
|
extra={"src": "AgentChat.Tools.EndMaintenance"})
|
|
return result
|
|
if not maintenance_id:
|
|
return "Error: maintenance_id is required unless end_all=true."
|
|
logger.reason("End maintenance event",
|
|
payload={"maintenance_id": maintenance_id},
|
|
extra={"src": "AgentChat.Tools.EndMaintenance"})
|
|
resp = await _post(f"/api/maintenance/{maintenance_id}/end")
|
|
result = _api_result(resp, {202})
|
|
logger.reflect("Maintenance ended", payload={"status": resp.status_code, "maintenance_id": maintenance_id},
|
|
extra={"src": "AgentChat.Tools.EndMaintenance"})
|
|
return result
|
|
# #endregion AgentChat.Tools.EndMaintenance
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# NEW: Agent-critical Superset operations (Phase 4)
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
# #region AgentChat.Tools.SupersetListDatabases [C:3] [TYPE Function] [SEMANTICS agent-chat,tools,superset,database,list]
|
|
# @ingroup AgentChat
|
|
# @BRIEF List all databases available in a Superset environment — returns id, name, uuid, engine.
|
|
# @POST Returns formatted string of database id → name (engine) per database.
|
|
# @SIDE_EFFECT HTTP GET to FastAPI /api/agent/superset/databases.
|
|
@tool
|
|
async def superset_list_databases(environment_id: str) -> str:
|
|
"""List all databases available in the current Superset environment. Returns database IDs and names."""
|
|
logger.reason("Listing Superset databases", payload={"environment_id": environment_id},
|
|
extra={"src": "AgentChat.Tools.SupersetListDatabases"})
|
|
try:
|
|
resp = await _get("/api/agent/superset/databases", params={"environment_id": environment_id})
|
|
if resp.status_code != 200:
|
|
return f"Error: failed to list databases (HTTP {resp.status_code})"
|
|
databases = resp.json()
|
|
if not databases:
|
|
return "No databases found in this environment."
|
|
lines = ["Available databases:"]
|
|
for db in databases:
|
|
db_id = db.get("id", "?")
|
|
db_name = db.get("database_name", db.get("name", "?"))
|
|
db_engine = db.get("backend", db.get("engine", ""))
|
|
if db_engine:
|
|
lines.append(f" • DB #{db_id}: {db_name} ({db_engine})")
|
|
else:
|
|
lines.append(f" • DB #{db_id}: {db_name}")
|
|
return "\n".join(lines)
|
|
except Exception as e:
|
|
logger.explore("Failed to list databases", error=str(e),
|
|
extra={"src": "AgentChat.Tools.SupersetListDatabases"})
|
|
return f"Error listing databases: {e}"
|
|
# #endregion AgentChat.Tools.SupersetListDatabases
|
|
|
|
|
|
# #region AgentChat.Tools.SupersetSql.Schema [C:1] [TYPE Class] [SEMANTICS agent-chat,tools,schema,superset,sql]
|
|
class ExecuteSupersetSqlInput(BaseModel):
|
|
environment_id: str = Field(description="Target Superset environment ID")
|
|
database_id: int = Field(description="Database connection ID")
|
|
sql: str = Field(description="SQL SELECT query to execute")
|
|
query_schema: str | None = Field(default=None, description="Optional schema for the query")
|
|
# #endregion AgentChat.Tools.SupersetSql.Schema
|
|
|
|
|
|
# #region AgentChat.Tools.SupersetSql [C:3] [TYPE Function] [SEMANTICS agent-chat,tools,superset,sql]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Execute a read-only SQL query in Superset SQL Lab (DDL/DML blocked).
|
|
# @PRE User authenticated via dual-identity JWT. Database exists in target environment.
|
|
# @POST Returns query results dict or error with blocked keyword.
|
|
# @SIDE_EFFECT HTTP POST to FastAPI /api/agent/superset/sqllab/execute.
|
|
@tool(args_schema=ExecuteSupersetSqlInput)
|
|
async def superset_execute_sql(environment_id: str, database_id: int, sql: str, query_schema: str | None = None) -> str:
|
|
"""Execute a read-only SQL query in Superset SQL Lab (SELECT only, DDL/DML blocked)."""
|
|
logger.reason("Execute Superset SQL", payload={"environment_id": environment_id, "database_id": database_id},
|
|
extra={"src": "AgentChat.Tools.SupersetSql"})
|
|
params: dict[str, Any] = {
|
|
"environment_id": environment_id,
|
|
"database_id": str(database_id),
|
|
"sql": sql,
|
|
}
|
|
if query_schema:
|
|
params["schema"] = query_schema
|
|
resp = await _post("/api/agent/superset/sqllab/execute", params=params)
|
|
result = _api_result(resp)
|
|
logger.reflect("Superset SQL executed", payload={"status": resp.status_code},
|
|
extra={"src": "AgentChat.Tools.SupersetSql"})
|
|
return result
|
|
# #endregion AgentChat.Tools.SupersetSql
|
|
|
|
|
|
# #region AgentChat.Tools.SupersetDatabases.Schema [C:1] [TYPE Class] [SEMANTICS agent-chat,tools,schema,superset,database]
|
|
class ListDatabaseSchemasInput(BaseModel):
|
|
environment_id: str = Field(description="Target Superset environment ID")
|
|
database_id: int = Field(description="Database connection ID")
|
|
schema_name: str | None = Field(default=None, description="Optional schema to list tables for")
|
|
table_name: str | None = Field(default=None, description="Optional table to get metadata/select_star for")
|
|
action: str = Field(default="schemas", description="Action: schemas, tables, table_metadata, select_star, validate_sql")
|
|
# #endregion AgentChat.Tools.SupersetDatabases.Schema
|
|
|
|
|
|
# #region AgentChat.Tools.SupersetDatabases [C:3] [TYPE Function] [SEMANTICS agent-chat,tools,superset,database,explore]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Explore database structure: list schemas, tables, get table metadata, generate SELECT *.
|
|
# @PRE User authenticated via dual-identity JWT. Database exists.
|
|
# @POST Returns schema/table/metadata info.
|
|
# @SIDE_EFFECT HTTP GET/POST to FastAPI /api/agent/superset/databases/{id}/...
|
|
@tool(args_schema=ListDatabaseSchemasInput)
|
|
async def superset_explore_database(
|
|
environment_id: str,
|
|
database_id: int,
|
|
schema_name: str | None = None,
|
|
table_name: str | None = None,
|
|
action: str = "schemas",
|
|
) -> str:
|
|
"""Explore database: list schemas, tables, get table metadata, or generate SELECT * template."""
|
|
logger.reason("Explore Superset database", payload={"environment_id": environment_id, "database_id": database_id, "action": action},
|
|
extra={"src": "AgentChat.Tools.SupersetDatabases"})
|
|
params: dict[str, Any] = {"environment_id": environment_id}
|
|
if action == "schemas":
|
|
path = f"/api/agent/superset/databases/{database_id}/schemas"
|
|
elif action == "tables":
|
|
if not schema_name:
|
|
return "Error: schema_name is required for tables action."
|
|
path = f"/api/agent/superset/databases/{database_id}/tables"
|
|
params["schema_name"] = schema_name
|
|
elif action == "table_metadata":
|
|
if not table_name:
|
|
return "Error: table_name is required for table_metadata action."
|
|
path = f"/api/agent/superset/databases/{database_id}/table_metadata"
|
|
params["table_name"] = table_name
|
|
if schema_name:
|
|
params["schema_name"] = schema_name
|
|
elif action == "select_star":
|
|
if not table_name:
|
|
return "Error: table_name is required for select_star action."
|
|
path = f"/api/agent/superset/databases/{database_id}/select_star"
|
|
params["table_name"] = table_name
|
|
if schema_name:
|
|
params["schema_name"] = schema_name
|
|
else:
|
|
return f"Error: unknown action '{action}'. Use: schemas, tables, table_metadata, select_star."
|
|
resp = await _get(path, params=params)
|
|
result = _api_result(resp)
|
|
logger.reflect("Database explored", payload={"status": resp.status_code, "action": action},
|
|
extra={"src": "AgentChat.Tools.SupersetDatabases"})
|
|
return result
|
|
# #endregion AgentChat.Tools.SupersetDatabases
|
|
|
|
|
|
# #region AgentChat.Tools.SupersetAudit.Schema [C:1] [TYPE Class] [SEMANTICS agent-chat,tools,schema,superset,audit]
|
|
class AuditPermissionsInput(BaseModel):
|
|
environment_id: str = Field(description="Target Superset environment ID")
|
|
username_filter: str | None = Field(default=None, description="Optional filter by username")
|
|
include_admin: bool = Field(default=False, description="Include Admin users in the audit")
|
|
# #endregion AgentChat.Tools.SupersetAudit.Schema
|
|
|
|
|
|
# #region AgentChat.Tools.SupersetAudit [C:3] [TYPE Function] [SEMANTICS agent-chat,tools,superset,audit,permissions]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Audit access rights: user x dashboard x dataset x RLS permission matrix.
|
|
# @PRE User authenticated via dual-identity JWT.
|
|
# @POST Returns audit report with per-user dashboard/dataset access and RLS regions.
|
|
# @SIDE_EFFECT HTTP GET to FastAPI /api/agent/superset/audit/permissions.
|
|
@tool(args_schema=AuditPermissionsInput)
|
|
async def superset_audit_permissions(
|
|
environment_id: str,
|
|
username_filter: str | None = None,
|
|
include_admin: bool = False,
|
|
) -> str:
|
|
"""Audit access rights: user x dashboard x dataset x RLS permission matrix."""
|
|
logger.reason("Audit Superset permissions", payload={"environment_id": environment_id},
|
|
extra={"src": "AgentChat.Tools.SupersetAudit"})
|
|
params: dict[str, Any] = {"environment_id": environment_id}
|
|
if username_filter:
|
|
params["username_filter"] = username_filter
|
|
if include_admin:
|
|
params["include_admin"] = "true"
|
|
resp = await _get("/api/agent/superset/audit/permissions", params=params)
|
|
result = _api_result(resp)
|
|
logger.reflect("Permissions audit completed", payload={"status": resp.status_code},
|
|
extra={"src": "AgentChat.Tools.SupersetAudit"})
|
|
return result
|
|
# #endregion AgentChat.Tools.SupersetAudit
|
|
|
|
|
|
# #region AgentChat.Tools.SupersetCreateDashboard.Schema [C:1] [TYPE Class] [SEMANTICS agent-chat,tools,schema,superset,dashboard,create]
|
|
class CreateSupersetDashboardInput(BaseModel):
|
|
environment_id: str = Field(description="Target Superset environment ID")
|
|
dashboard_title: str = Field(description="Dashboard title")
|
|
slug: str | None = Field(default=None, description="Optional URL slug")
|
|
published: bool = Field(default=False, description="Whether the dashboard is published")
|
|
# #endregion AgentChat.Tools.SupersetCreateDashboard.Schema
|
|
|
|
|
|
# #region AgentChat.Tools.SupersetCreateDashboard [C:3] [TYPE Function] [SEMANTICS agent-chat,tools,superset,dashboard,create]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Create a new dashboard in Superset.
|
|
# @PRE User authenticated via dual-identity JWT.
|
|
# @POST Returns created dashboard dict.
|
|
# @SIDE_EFFECT HTTP POST to FastAPI /api/agent/superset/dashboards.
|
|
@tool(args_schema=CreateSupersetDashboardInput)
|
|
async def superset_create_dashboard(
|
|
environment_id: str,
|
|
dashboard_title: str,
|
|
slug: str | None = None,
|
|
published: bool = False,
|
|
) -> str:
|
|
"""Create a new dashboard in Superset."""
|
|
logger.reason("Create Superset dashboard", payload={"title": dashboard_title, "environment_id": environment_id},
|
|
extra={"src": "AgentChat.Tools.SupersetCreateDashboard"})
|
|
params: dict[str, Any] = {
|
|
"environment_id": environment_id,
|
|
"dashboard_title": dashboard_title,
|
|
"published": "true" if published else "false",
|
|
}
|
|
if slug:
|
|
params["slug"] = slug
|
|
resp = await _post("/api/agent/superset/dashboards", params=params)
|
|
result = _api_result(resp, {200, 201})
|
|
logger.reflect("Dashboard created", payload={"status": resp.status_code},
|
|
extra={"src": "AgentChat.Tools.SupersetCreateDashboard"})
|
|
return result
|
|
# #endregion AgentChat.Tools.SupersetCreateDashboard
|
|
|
|
|
|
# #region AgentChat.Tools.SupersetCopyDashboard.Schema [C:1] [TYPE Class] [SEMANTICS agent-chat,tools,schema,superset,dashboard,copy]
|
|
class CopySupersetDashboardInput(BaseModel):
|
|
environment_id: str = Field(description="Target Superset environment ID")
|
|
dashboard_id: int = Field(description="Source dashboard ID to copy")
|
|
dashboard_title: str | None = Field(default=None, description="Optional title for the copy")
|
|
# #endregion AgentChat.Tools.SupersetCopyDashboard.Schema
|
|
|
|
|
|
# #region AgentChat.Tools.SupersetCopyDashboard [C:3] [TYPE Function] [SEMANTICS agent-chat,tools,superset,dashboard,copy]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Deep-copy a Superset dashboard including all charts.
|
|
# @PRE User authenticated via dual-identity JWT.
|
|
# @POST Returns copy result dict.
|
|
# @SIDE_EFFECT HTTP POST to FastAPI /api/agent/superset/dashboards/{id}/copy.
|
|
@tool(args_schema=CopySupersetDashboardInput)
|
|
async def superset_copy_dashboard(
|
|
environment_id: str,
|
|
dashboard_id: int,
|
|
dashboard_title: str | None = None,
|
|
) -> str:
|
|
"""Deep-copy a dashboard including all charts."""
|
|
logger.reason("Copy Superset dashboard", payload={"dashboard_id": dashboard_id, "environment_id": environment_id},
|
|
extra={"src": "AgentChat.Tools.SupersetCopyDashboard"})
|
|
params: dict[str, Any] = {"environment_id": environment_id}
|
|
if dashboard_title:
|
|
params["dashboard_title"] = dashboard_title
|
|
resp = await _post(f"/api/agent/superset/dashboards/{dashboard_id}/copy", params=params)
|
|
result = _api_result(resp, {200, 201})
|
|
logger.reflect("Dashboard copied", payload={"status": resp.status_code, "source_id": dashboard_id},
|
|
extra={"src": "AgentChat.Tools.SupersetCopyDashboard"})
|
|
return result
|
|
# #endregion AgentChat.Tools.SupersetCopyDashboard
|
|
|
|
|
|
# #region AgentChat.Tools.SupersetCreateDataset.Schema [C:1] [TYPE Class] [SEMANTICS agent-chat,tools,schema,superset,dataset,create]
|
|
class CreateSupersetDatasetInput(BaseModel):
|
|
environment_id: str = Field(description="Target Superset environment ID")
|
|
table_name: str = Field(description="Table or view name")
|
|
database: int = Field(description="Database connection ID")
|
|
schema_name: str | None = Field(default=None, description="Optional schema name")
|
|
# #endregion AgentChat.Tools.SupersetCreateDataset.Schema
|
|
|
|
|
|
# #region AgentChat.Tools.SupersetCreateDataset [C:3] [TYPE Function] [SEMANTICS agent-chat,tools,superset,dataset,create]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Create a new dataset in Superset.
|
|
# @PRE User authenticated via dual-identity JWT.
|
|
# @POST Returns created dataset dict.
|
|
# @SIDE_EFFECT HTTP POST to FastAPI /api/agent/superset/datasets.
|
|
@tool(args_schema=CreateSupersetDatasetInput)
|
|
async def superset_create_dataset(
|
|
environment_id: str,
|
|
table_name: str,
|
|
database: int,
|
|
schema_name: str | None = None,
|
|
) -> str:
|
|
"""Create a new dataset in Superset."""
|
|
logger.reason("Create Superset dataset", payload={"table_name": table_name, "environment_id": environment_id},
|
|
extra={"src": "AgentChat.Tools.SupersetCreateDataset"})
|
|
params: dict[str, Any] = {
|
|
"environment_id": environment_id,
|
|
"table_name": table_name,
|
|
"database": str(database),
|
|
}
|
|
if schema_name:
|
|
params["schema_name"] = schema_name
|
|
resp = await _post("/api/agent/superset/datasets", params=params)
|
|
result = _api_result(resp, {200, 201})
|
|
logger.reflect("Dataset created", payload={"status": resp.status_code},
|
|
extra={"src": "AgentChat.Tools.SupersetCreateDataset"})
|
|
return result
|
|
# #endregion AgentChat.Tools.SupersetCreateDataset
|
|
|
|
|
|
# #region AgentChat.Tools.SupersetFormatSql [C:2] [TYPE Function] [SEMANTICS agent-chat,tools,superset,sql,format]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Format/pretty-print a SQL query using Superset SQL Lab.
|
|
# @SIDE_EFFECT HTTP POST to FastAPI /api/agent/superset/sqllab/format.
|
|
@tool
|
|
async def superset_format_sql(environment_id: str, sql: str) -> str:
|
|
"""Format/pretty-print a SQL query using Superset SQL Lab."""
|
|
logger.reason("Format SQL", extra={"src": "AgentChat.Tools.SupersetFormatSql"})
|
|
resp = await _post("/api/agent/superset/sqllab/format", params={
|
|
"environment_id": environment_id,
|
|
"sql": sql,
|
|
})
|
|
result = _api_result(resp)
|
|
logger.reflect("SQL formatted", extra={"src": "AgentChat.Tools.SupersetFormatSql"})
|
|
return result
|
|
# #endregion AgentChat.Tools.SupersetFormatSql
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# Tool registry
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
# #region AgentChat.Tools.GetAll [C:2] [TYPE Function] [SEMANTICS agent-chat,tools,registry]
|
|
# @ingroup AgentChat
|
|
# @BRIEF Return the full list of registered LangChain @tool functions.
|
|
def get_all_tools() -> list:
|
|
return [
|
|
show_capabilities,
|
|
search_dashboards,
|
|
get_health_summary,
|
|
list_environments,
|
|
get_task_status,
|
|
list_llm_providers,
|
|
get_llm_status,
|
|
create_branch,
|
|
commit_changes,
|
|
deploy_dashboard,
|
|
execute_migration,
|
|
run_backup,
|
|
run_llm_validation,
|
|
run_llm_documentation,
|
|
list_maintenance_events,
|
|
start_maintenance,
|
|
end_maintenance,
|
|
# NEW: Superset direct operations
|
|
superset_list_databases,
|
|
superset_execute_sql,
|
|
superset_explore_database,
|
|
superset_audit_permissions,
|
|
superset_create_dashboard,
|
|
superset_copy_dashboard,
|
|
superset_create_dataset,
|
|
superset_format_sql,
|
|
]
|
|
# #endregion AgentChat.Tools.GetAll
|
|
|
|
# ── Optional overrides for embedding descriptions (auto-generated from docstring) ──
|
|
_TOOL_DESCRIPTIONS_OVERRIDES: dict[str, str] = {}
|
|
|
|
# #endregion AgentChat.Tools
|