"""HTTP API for the AI layer. * ``GET /api/v1/ai/semantic/`` — list registered entities * ``GET /api/v1/ai/semantic//`` — entity detail * ``GET /api/v1/ai/actions/`` — list registered actions * ``POST /api/v1/ai/actions//execute/`` — execute (audit-logged) * ``GET /api/v1/ai/suggestions/`` — list suggestions * ``POST /api/v1/ai/suggestions/`` — create suggestion * ``POST /api/v1/ai/suggestions//decide/`` — approve / reject * ``GET /api/v1/ai/agents/`` — list registered agents * ``POST /api/v1/ai/agents//run/`` — run one chat round """ from __future__ import annotations from rest_framework import status from rest_framework.exceptions import NotFound, ValidationError from rest_framework.permissions import IsAuthenticated from rest_framework.request import Request from rest_framework.response import Response from rest_framework.views import APIView from simorgh.apps.ai.actions import ActionError, get_action, list_actions, serialize_action from simorgh.apps.ai.agents import AgentError, get_agent, list_agents, run_agent from simorgh.apps.ai.context import ContextError, build_context from simorgh.apps.ai.models import AISuggestion from simorgh.apps.ai.semantic import ( SemanticError, get_entity, list_entities, serialize_entity, ) from simorgh.apps.ai.services import ( AIServiceError, create_suggestion, decide_suggestion, execute_action, list_suggestions, ) from simorgh.apps.iam.permissions import HasPermission from simorgh.core.context import current_request_context def _require_tenant(): ctx = current_request_context() if ctx.tenant is None: raise ValidationError({"detail": "tenant required (set X-Tenant header)"}) return ctx.tenant def _suggestion_payload(s: AISuggestion) -> dict: return { "id": str(s.public_id), "kind": s.kind, "status": s.status, "title": s.title, "summary": s.summary, "source": s.source, "proposed_action": s.proposed_action, "payload": s.payload, "metadata": s.metadata, "created_at": s.created_at.isoformat(), } # --------------------------------------------------------------------------- # Semantic registry # --------------------------------------------------------------------------- class SemanticListView(APIView): permission_classes = (IsAuthenticated, HasPermission) required_permission = "ai.semantic.view" def get(self, _request: Request) -> Response: _require_tenant() return Response({"results": [serialize_entity(e) for e in list_entities()]}) class SemanticDetailView(APIView): permission_classes = (IsAuthenticated, HasPermission) required_permission = "ai.semantic.view" def get(self, _request: Request, key: str) -> Response: _require_tenant() try: return Response(serialize_entity(get_entity(key))) except SemanticError as exc: raise NotFound(str(exc)) from exc # --------------------------------------------------------------------------- # Actions # --------------------------------------------------------------------------- class ActionListView(APIView): permission_classes = (IsAuthenticated, HasPermission) required_permission = "ai.semantic.view" def get(self, _request: Request) -> Response: _require_tenant() return Response({"results": [serialize_action(a) for a in list_actions()]}) class ActionExecuteView(APIView): permission_classes = (IsAuthenticated, HasPermission) required_permission = "ai.action.execute" def post(self, request: Request, key: str) -> Response: tenant = _require_tenant() try: get_action(key) except ActionError as exc: raise NotFound(str(exc)) from exc payload = request.data.get("payload") or {} suggestion_id = request.data.get("suggestion_id") try: log = execute_action( tenant, key, payload=payload, actor_id=request.user.pk if request.user.is_authenticated else None, suggestion_id=suggestion_id, ) except (ActionError, AIServiceError) as exc: raise ValidationError({"detail": str(exc)}) from exc return Response( { "id": str(log.public_id), "action": log.action_key, "status": log.status, "result": log.result, "error": log.error, }, status=status.HTTP_200_OK, ) # --------------------------------------------------------------------------- # Suggestions # --------------------------------------------------------------------------- class SuggestionListCreateView(APIView): permission_classes = (IsAuthenticated, HasPermission) def get_required_permission(self, request: Request, _view) -> str: return "ai.audit.view" if request.method == "GET" else "ai.action.execute" def get(self, request: Request) -> Response: tenant = _require_tenant() items = list_suggestions( tenant, status=request.query_params.get("status"), kind=request.query_params.get("kind"), ) return Response({"results": [_suggestion_payload(s) for s in items]}) def post(self, request: Request) -> Response: tenant = _require_tenant() title = request.data.get("title") if not title: raise ValidationError({"title": "required"}) try: s = create_suggestion( tenant, title=title, kind=request.data.get("kind", "agent_output"), summary=request.data.get("summary", ""), source=request.data.get("source", ""), proposed_action=request.data.get("proposed_action", ""), payload=request.data.get("payload") or {}, metadata=request.data.get("metadata") or {}, created_by_id=request.user.pk if request.user.is_authenticated else None, ) except (ActionError, AIServiceError) as exc: raise ValidationError({"detail": str(exc)}) from exc return Response(_suggestion_payload(s), status=status.HTTP_201_CREATED) class SuggestionDecideView(APIView): permission_classes = (IsAuthenticated, HasPermission) required_permission = "ai.action.approve" def post(self, request: Request, suggestion_id: str) -> Response: tenant = _require_tenant() decision = request.data.get("decision") if not decision: raise ValidationError({"decision": "required"}) try: approval = decide_suggestion( tenant, suggestion_id, decision=decision, actor_id=request.user.pk if request.user.is_authenticated else None, note=request.data.get("note", ""), ) except AIServiceError as exc: raise ValidationError({"detail": str(exc)}) from exc return Response( { "id": str(approval.public_id), "suggestion_id": str(approval.suggestion_id), "decision": approval.decision, "note": approval.note, }, status=status.HTTP_200_OK, ) # --------------------------------------------------------------------------- # Agents # --------------------------------------------------------------------------- class AgentListView(APIView): permission_classes = (IsAuthenticated, HasPermission) required_permission = "ai.agent.run" def get(self, _request: Request) -> Response: _require_tenant() return Response( { "results": [ { "key": a.key, "label_key": a.label_key, "provider": a.provider, "description": a.description, "allowed_tools": list(a.allowed_tools), } for a in list_agents() ], }, ) class AgentRunView(APIView): permission_classes = (IsAuthenticated, HasPermission) required_permission = "ai.agent.run" def post(self, request: Request, key: str) -> Response: tenant = _require_tenant() try: get_agent(key) except AgentError as exc: raise NotFound(str(exc)) from exc user_input = request.data.get("input") or "" if not user_input: raise ValidationError({"input": "required"}) entity_keys = request.data.get("entity_keys") or [] try: context = build_context( tenant=tenant, entity_keys=entity_keys, recall_query=request.data.get("recall_query", ""), recall_collection=request.data.get("recall_collection", ""), ) except (ContextError, SemanticError) as exc: raise ValidationError({"detail": str(exc)}) from exc try: run = run_agent( key, user_input=user_input, context=context, tool_calls=request.data.get("tool_calls") or [], ) except AgentError as exc: raise ValidationError({"detail": str(exc)}) from exc return Response( { "agent": run.agent, "response": { "content": run.response.content, "provider": run.response.provider, "model": run.response.model, "usage": run.response.usage, }, "tool_invocations": run.tool_invocations, "context": context.to_dict(), }, )