"""API views for the agency app."""
import requests
import django
from django.conf import settings
from django.contrib.auth import get_user_model
from django.contrib.auth.models import Group, Permission
from django.shortcuts import render
from importlib.metadata import version as get_package_version
from rest_framework import (
permissions, views, viewsets, serializers, status, mixins
)
from rest_framework.response import Response
from rest_framework.decorators import api_view, permission_classes, action
from drf_spectacular.utils import extend_schema, extend_schema_view
from drf_spectacular.openapi import AutoSchema
from agency.api.permissions import (
filter_by_owner,
filter_by_tenant_membership,
filter_documents_by_tenant_membership,
)
from agency.models import Agent, Mission, Execution, Thread, Message, Tool, Model, Provider, UserPreferences, Task, Voice
User = get_user_model()
[docs]
def serializer(cls):
"""Get the default model serializer.
Args:
cls: Model class to create serializer for
Returns:
ModelSerializer class for the given model
"""
class Serializer(serializers.ModelSerializer):
class Meta:
model = cls
fields = '__all__'
Serializer.__name__ = cls.__name__ + "Serializer"
return Serializer
[docs]
def provider_serializer():
"""Provider row plus key/sync hints (never the secret itself)."""
from agency.providers import is_syncable, key_configured
class ProviderSerializer(serializers.ModelSerializer):
key_configured = serializers.SerializerMethodField()
syncable = serializers.SerializerMethodField()
class Meta:
model = Provider
fields = "__all__"
def get_key_configured(self, obj):
return key_configured(obj)
def get_syncable(self, obj):
return is_syncable(obj)
return ProviderSerializer
[docs]
def model_serializer():
"""Model row with temperature / num_ctx mirrored from ``metadata``."""
_MISSING = object()
class ModelSerializer(serializers.ModelSerializer):
temperature = serializers.FloatField(required=False, allow_null=True)
num_ctx = serializers.IntegerField(required=False, allow_null=True)
class Meta:
model = Model
fields = "__all__"
def to_representation(self, instance):
data = super().to_representation(instance)
meta = instance.metadata or {}
if "temperature" in meta:
data["temperature"] = meta["temperature"]
if "num_ctx" in meta:
data["num_ctx"] = meta["num_ctx"]
return data
def _pop_runtime(self, validated_data):
return (
validated_data.pop("temperature", _MISSING),
validated_data.pop("num_ctx", _MISSING),
)
def _write_runtime(self, instance, temperature, num_ctx):
if temperature is _MISSING and num_ctx is _MISSING:
return False
meta = dict(instance.metadata or {})
if temperature is not _MISSING:
if temperature is None:
meta.pop("temperature", None)
else:
meta["temperature"] = temperature
if num_ctx is not _MISSING:
if num_ctx is None:
meta.pop("num_ctx", None)
else:
meta["num_ctx"] = int(num_ctx)
instance.metadata = meta or None
return True
def create(self, validated_data):
temperature, num_ctx = self._pop_runtime(validated_data)
instance = super().create(validated_data)
if self._write_runtime(instance, temperature, num_ctx):
instance.save(update_fields=["metadata"])
return instance
def update(self, instance, validated_data):
temperature, num_ctx = self._pop_runtime(validated_data)
instance = super().update(instance, validated_data)
if self._write_runtime(instance, temperature, num_ctx):
instance.save(update_fields=["metadata"])
return instance
return ModelSerializer
[docs]
def agent_serializer():
"""Agent serializer with nested model/tools for explorers and M2M write."""
class ToolBriefSerializer(serializers.ModelSerializer):
icon = serializers.SerializerMethodField()
class Meta:
model = Tool
fields = ("id", "definition", "help", "icon")
def get_icon(self, obj):
if not obj.icon:
return None
try:
url = obj.icon.url
except ValueError:
return None
request = self.context.get("request")
if request is not None:
return request.build_absolute_uri(url)
return url
class ModelBriefSerializer(serializers.ModelSerializer):
class Meta:
model = Model
fields = ("id", "name")
class VoiceBriefSerializer(serializers.ModelSerializer):
class Meta:
model = Voice
fields = ("id", "name", "provider", "key", "language")
class AgentSerializer(serializers.ModelSerializer):
model = ModelBriefSerializer(read_only=True)
model_id = serializers.PrimaryKeyRelatedField(
queryset=Model.objects.all(),
source="model",
write_only=False,
)
voice = VoiceBriefSerializer(read_only=True)
voice_id = serializers.PrimaryKeyRelatedField(
queryset=Voice.objects.all(),
source="voice",
write_only=False,
required=False,
allow_null=True,
)
tools = ToolBriefSerializer(many=True, read_only=True)
tool_ids = serializers.PrimaryKeyRelatedField(
many=True,
queryset=Tool.objects.all(),
source="tools",
write_only=True,
required=False,
)
tools_count = serializers.SerializerMethodField()
icon = serializers.ImageField(required=False, allow_null=True)
class Meta:
model = Agent
fields = (
"id",
"name",
"namespace",
"system_key",
"is_customized",
"instructions",
"model",
"model_id",
"voice",
"voice_id",
"tools",
"tool_ids",
"tools_count",
"deep",
"icon",
"created_at",
"updated_at",
)
read_only_fields = (
"id",
"system_key",
"is_customized",
"created_at",
"updated_at",
"tools_count",
)
def validate_namespace(self, value):
from agency.namespaces import normalize_submitted_namespace
try:
return normalize_submitted_namespace(value)
except ValueError as exc:
raise serializers.ValidationError(str(exc)) from exc
def get_tools_count(self, obj):
if hasattr(obj, "_prefetched_objects_cache") and "tools" in obj._prefetched_objects_cache:
return len(obj.tools.all())
return obj.tools.count()
def to_representation(self, instance):
data = super().to_representation(instance)
if instance.icon:
try:
url = instance.icon.url
except ValueError:
data["icon"] = None
else:
request = self.context.get("request")
data["icon"] = (
request.build_absolute_uri(url) if request is not None else url
)
else:
data["icon"] = None
return data
def create(self, validated_data):
tools = validated_data.pop("tools", None)
from agency.namespaces import infer_agent_namespace
if not validated_data.get("namespace"):
validated_data["namespace"] = infer_agent_namespace(
name=validated_data.get("name", ""),
system_key=validated_data.get("system_key"),
)
agent = Agent.objects.create(**validated_data)
if tools is not None:
agent.tools.set(tools)
return agent
def update(self, instance, validated_data):
tools = validated_data.pop("tools", None)
if instance.system_key and "instructions" in validated_data:
if validated_data["instructions"] != instance.instructions:
validated_data["is_customized"] = True
if "namespace" in validated_data and not validated_data["namespace"]:
from agency.namespaces import infer_agent_namespace
validated_data["namespace"] = infer_agent_namespace(
name=validated_data.get("name", instance.name),
system_key=instance.system_key,
)
for attr, value in validated_data.items():
setattr(instance, attr, value)
instance.save()
if tools is not None:
instance.tools.set(tools)
return instance
AgentSerializer.__name__ = "AgentSerializer"
return AgentSerializer
[docs]
def agent_view_set():
"""CRUD viewset for Agent with prefetch for explorer cards."""
from rest_framework.parsers import FormParser, JSONParser, MultiPartParser
ser = agent_serializer()
base = view_set(Agent, True, serializer_class=ser)
class AgentViewSet(base):
parser_classes = [JSONParser, MultiPartParser, FormParser]
def get_queryset(self):
return (
Agent.objects.all()
.select_related("model", "voice")
.prefetch_related("tools")
.order_by("name")
)
AgentViewSet.__name__ = "AgentViewSet"
return AgentViewSet
[docs]
def mission_serializer():
"""Mission serializer with loop fields and read-only is_loop."""
class MissionSerializer(serializers.ModelSerializer):
is_loop = serializers.BooleanField(read_only=True)
class Meta:
model = Mission
fields = (
"id",
"name",
"namespace",
"system_key",
"is_customized",
"prompt",
"loop_prompt",
"max_iterations",
"loop_delay_seconds",
"is_loop",
"created_at",
"updated_at",
)
read_only_fields = (
"id",
"system_key",
"is_customized",
"created_at",
"updated_at",
"is_loop",
)
def validate_namespace(self, value):
from agency.namespaces import normalize_submitted_namespace
try:
return normalize_submitted_namespace(value)
except ValueError as exc:
raise serializers.ValidationError(str(exc)) from exc
def validate(self, attrs):
from agency.mission.loop import normalize_loop_prompt, validate_loop_limits
loop_prompt = attrs.get("loop_prompt", getattr(self.instance, "loop_prompt", None))
loop_prompt = normalize_loop_prompt(loop_prompt)
attrs["loop_prompt"] = loop_prompt
max_iterations = attrs.get(
"max_iterations",
getattr(self.instance, "max_iterations", 10),
)
loop_delay = attrs.get(
"loop_delay_seconds",
getattr(self.instance, "loop_delay_seconds", 0),
)
if loop_prompt is not None:
validate_loop_limits(
max_iterations=max_iterations,
loop_delay_seconds=loop_delay,
)
from agency.namespaces import infer_mission_namespace
name = attrs.get("name", getattr(self.instance, "name", ""))
if "namespace" in attrs:
attrs["namespace"] = attrs["namespace"] or infer_mission_namespace(
name=name,
system_key=getattr(self.instance, "system_key", None),
)
elif not getattr(self.instance, "namespace", ""):
attrs["namespace"] = infer_mission_namespace(
name=name,
system_key=getattr(self.instance, "system_key", None),
)
return attrs
def update(self, instance, validated_data):
content_keys = {"prompt", "loop_prompt", "max_iterations", "loop_delay_seconds"}
if instance.system_key and content_keys.intersection(validated_data):
for key in content_keys:
if key in validated_data and getattr(instance, key) != validated_data[key]:
validated_data["is_customized"] = True
break
return super().update(instance, validated_data)
MissionSerializer.__name__ = "MissionSerializer"
return MissionSerializer
[docs]
def mission_view_set():
ser = mission_serializer()
base = view_set(Mission, True, serializer_class=ser)
class MissionViewSet(base):
def get_queryset(self):
return Mission.objects.all().order_by("name")
MissionViewSet.__name__ = "MissionViewSet"
return MissionViewSet
[docs]
def execution_serializer():
"""List/detail serializers for Execution (Mission Control explorers)."""
class ExecutionListSerializer(serializers.ModelSerializer):
mission_name = serializers.CharField(source="mission.name", read_only=True)
agent_name = serializers.CharField(source="agent.name", read_only=True)
has_result = serializers.SerializerMethodField()
is_loop = serializers.SerializerMethodField()
class Meta:
model = Execution
fields = (
"id",
"user",
"agent",
"agent_name",
"mission",
"mission_name",
"status",
"has_result",
"iteration",
"max_iterations",
"loop_status",
"next_run_at",
"is_loop",
"created_at",
"updated_at",
)
def get_has_result(self, obj):
result = obj.result or {}
return bool(isinstance(result, dict) and result.get("messages") is not None)
def get_is_loop(self, obj):
from agency.mission.loop import get_loop_config
return bool(get_loop_config(obj).get("is_loop"))
class ExecutionDetailSerializer(ExecutionListSerializer):
messages = serializers.SerializerMethodField()
thread_id = serializers.SerializerMethodField()
interruptions = serializers.SerializerMethodField()
threads = serializers.SerializerMethodField()
result_messages = serializers.SerializerMethodField()
class Meta(ExecutionListSerializer.Meta):
fields = ExecutionListSerializer.Meta.fields + (
"result",
"metadata",
"loop_decision",
"messages",
"result_messages",
"thread_id",
"interruptions",
"threads",
)
def get_messages(self, obj):
thread = obj.threads.first()
if not thread:
return []
messages = (
Message.objects.filter(checkpoint__thread=thread)
.select_related("checkpoint", "agent", "user")
.order_by("created_at")
)
from agency.utils import serializable
return [
serializable(
{
"id": msg.id,
"role": msg.role,
"content": msg.content,
"think": msg.think,
"tool_calls": msg.tool_calls,
"metadata": msg.metadata,
"created_at": msg.created_at.isoformat() if msg.created_at else None,
"agent_id": str(msg.agent.id) if msg.agent else None,
"agent_name": msg.agent.name if msg.agent else None,
}
)
for msg in messages
]
def get_result_messages(self, obj):
"""LangGraph messages from ``execution.result`` (admin Result source)."""
result = obj.result or {}
if not isinstance(result, dict):
return []
msgs = result.get("messages")
return msgs if isinstance(msgs, list) else []
def get_thread_id(self, obj):
thread = obj.threads.first()
return str(thread.id) if thread else None
def get_interruptions(self, obj):
from agency.models import Interruption
rows = (
Interruption.objects.filter(checkpoint__run__thread__execution=obj)
.select_related(
"checkpoint",
"checkpoint__run",
"checkpoint__run__thread",
"user",
"agent",
)
.order_by("created_at")
)
out = []
for row in rows:
try:
from_label = row.get_from()
except Exception:
from_label = ""
try:
to_label = row.get_to()
except Exception:
to_label = ""
out.append(
{
"id": str(row.id),
"status": row.status,
"priority": row.priority,
"checkpoint_id": row.checkpoint_id,
"checkpoint_pk": row.checkpoint.id if row.checkpoint_id else None,
"question": row.question or {},
"answer": row.answer or {},
"from": from_label,
"to": to_label,
"agent_name": row.agent.name if row.agent_id else None,
"user_name": row.user.username if row.user_id else None,
"created_at": row.created_at.isoformat() if row.created_at else None,
"answered_at": row.answered_at.isoformat() if row.answered_at else None,
"resumed_at": row.resumed_at.isoformat() if row.resumed_at else None,
}
)
return out
def get_threads(self, obj):
threads = (
obj.threads.select_related("user", "agent")
.prefetch_related("runs__checkpoints")
.order_by("created_at")
)
out = []
for thread in threads:
runs_out = []
for run in thread.runs.all().order_by("id"):
cps = []
for cp in run.checkpoints.all().order_by("id"):
ts = getattr(cp, "ts", None)
if ts is not None and hasattr(ts, "isoformat"):
ts_s = ts.isoformat()
else:
ts_s = str(ts or "")
cps.append(
{
"id": cp.id,
"checkpoint_id": cp.checkpoint_id,
"namespace": cp.namespace,
"ts": ts_s,
"created_at": cp.created_at.isoformat() if cp.created_at else None,
"type": getattr(cp, "type", None),
}
)
runs_out.append(
{
"id": run.id,
"status": run.status,
"total_tokens": run.total_tokens,
"error": run.error or {},
"created_at": run.created_at.isoformat() if run.created_at else None,
"updated_at": run.updated_at.isoformat() if run.updated_at else None,
"checkpoints": cps,
}
)
out.append(
{
"id": str(thread.id),
"title": thread.title,
"is_active": thread.is_active,
"user_name": thread.user.username if thread.user_id else None,
"agent_name": thread.agent.name if thread.agent_id else None,
"created_at": thread.created_at.isoformat() if thread.created_at else None,
"runs": runs_out,
}
)
return out
ExecutionListSerializer.__name__ = "ExecutionListSerializer"
ExecutionDetailSerializer.__name__ = "ExecutionDetailSerializer"
return ExecutionListSerializer, ExecutionDetailSerializer
[docs]
def task_view_set():
"""List/retrieve Celery tasks (newest first for the UI monitor)."""
base = view_set(Task, False)
class TaskViewSet(base):
def get_queryset(self):
return Task.objects.all().order_by('-created_at')
TaskViewSet.__name__ = 'TaskViewSet'
return TaskViewSet
[docs]
def execution_view_set():
"""List/retrieve Execution with nested Result payload on retrieve."""
list_ser, detail_ser = execution_serializer()
class ExecutionViewSet(
mixins.ListModelMixin,
mixins.RetrieveModelMixin,
viewsets.GenericViewSet,
):
permission_classes = [permissions.IsAuthenticated]
def get_queryset(self):
qs = filter_by_owner(
Execution.objects.select_related("mission", "agent", "user"),
self.request.user,
).order_by("-updated_at")
mission = self.request.query_params.get("mission")
if mission:
qs = qs.filter(mission_id=mission)
agent = self.request.query_params.get("agent")
if agent:
qs = qs.filter(agent_id=agent)
status_filter = self.request.query_params.get("status")
if status_filter:
qs = qs.filter(status=status_filter)
return qs
def get_serializer_class(self):
if self.action == "retrieve":
return detail_ser
return list_ser
@extend_schema(
summary="Continue mission chat on this execution thread",
request={
"application/json": {
"type": "object",
"properties": {"message": {"type": "string"}},
"required": ["message"],
}
},
)
@action(
detail=True,
methods=["post"],
url_path="message",
permission_classes=[permissions.IsAuthenticated],
)
def post_message(self, request, pk=None):
"""Queue a follow-up user turn on the execution's Thread (Celery)."""
execution = self.get_object()
message = ""
if isinstance(request.data, dict):
message = str(request.data.get("message") or "").strip()
if not message:
return Response(
{"detail": "message is required"},
status=status.HTTP_400_BAD_REQUEST,
)
blocked = {
Execution.Status.STARTING,
Execution.Status.INTERRUPTED,
}
from agency.mission.loop import get_loop_config
is_loop = bool(get_loop_config(execution).get("is_loop"))
if not is_loop:
blocked.add(Execution.Status.RUNNING)
if execution.status in blocked:
return Response(
{
"detail": (
f"Cannot chat while execution is {execution.status}. "
"Answer pending HITL via the channel overlay if interrupted."
),
"status": execution.status,
},
status=status.HTTP_409_CONFLICT,
)
from agency.tasks import continue_mission
task = continue_mission.delay(
str(execution.id), message, request.user.pk,
)
thread = execution.threads.order_by("created_at").first()
return Response(
{
"execution_id": str(execution.id),
"thread_id": str(thread.id) if thread else None,
"task_id": str(task.id) if task else None,
"status": "queued",
},
status=status.HTTP_202_ACCEPTED,
)
ExecutionViewSet.__name__ = "ExecutionViewSet"
return ExecutionViewSet
[docs]
@extend_schema(tags=['system'])
class SystemInfo(views.APIView):
"""Host versions plus inference (Ollama GPUs, ComfyUI) when deployed."""
permission_classes = [permissions.IsAuthenticated]
schema = AutoSchema()
[docs]
@extend_schema(summary="Host and inference runtime info")
def get(self, request):
from agency.hostinfo import collect_system_info
return Response(collect_system_info(), status=status.HTTP_200_OK)
[docs]
class Version(views.APIView):
"""Returns version and system information."""
schema = AutoSchema()
[docs]
@extend_schema(
summary="Agency system versions",
)
def get(self, request):
models = requests.get(f"{settings.OLLAMA_URL}/api/ps")
collections = requests.get(f"{settings.CHROMADB_URL}/api/v1/collections/")
return Response({
"agency": settings.VERSION,
"django": django.__version__,
"langchain": get_package_version('langchain'),
'langchain_ollama': get_package_version('langchain-ollama'),
"langgraph": get_package_version('langgraph'),
"langchain-text-splitters": get_package_version('langchain-text-splitters'),
"numpy": get_package_version('numpy'),
"huggingface-hub": get_package_version('huggingface-hub'),
"collections": collections.json(),
"llm": models.json(),
"user": request.user.id
}, status=status.HTTP_200_OK)
[docs]
class MePrefs(views.APIView):
"""Current user's Mission Control preferences (workspace + Desktop)."""
permission_classes = [permissions.IsAuthenticated]
schema = AutoSchema()
[docs]
@extend_schema(summary="Get user preferences")
def get(self, request):
prefs = UserPreferences.for_user(request.user)
return Response(prefs.to_api())
[docs]
@extend_schema(summary="Update user preferences")
def patch(self, request):
prefs = UserPreferences.for_user(request.user)
data = request.data if isinstance(request.data, dict) else {}
update_fields: list[str] = []
if "workspaceId" in data:
raw = data.get("workspaceId")
if raw in (None, ""):
prefs.active_workspace = None
else:
from agency.corpus.models import Workspace
try:
prefs.active_workspace = Workspace.objects.get(pk=raw)
except (Workspace.DoesNotExist, ValueError, TypeError):
return Response(
{"detail": "workspaceId not found"},
status=status.HTTP_400_BAD_REQUEST,
)
update_fields.append("active_workspace")
if "desktop" in data:
desktop = data.get("desktop")
if desktop is None:
prefs.desktop = {}
elif isinstance(desktop, dict):
from agency.media import preserve_now_playing
prefs.desktop = preserve_now_playing(prefs.desktop, desktop)
else:
return Response(
{"detail": "desktop must be an object"},
status=status.HTTP_400_BAD_REQUEST,
)
update_fields.append("desktop")
if "language" in data:
from agency.ui.language import normalize_language
raw = data.get("language")
if raw in (None, ""):
prefs.language = ""
else:
prefs.language = normalize_language(str(raw))
update_fields.append("language")
if "timezone" in data:
from agency.ui.timezone import normalize_timezone
raw = data.get("timezone")
if raw in (None, ""):
prefs.timezone = ""
else:
try:
prefs.timezone = normalize_timezone(str(raw))
except ValueError:
return Response(
{"detail": "invalid timezone"},
status=status.HTTP_400_BAD_REQUEST,
)
update_fields.append("timezone")
if "systemVoiceId" in data or "ttsVoice" in data:
from agency.voices import get_voice
if "systemVoiceId" in data:
raw = data.get("systemVoiceId")
if raw in (None, ""):
prefs.system_voice = None
else:
try:
prefs.system_voice = Voice.objects.get(pk=raw)
except (Voice.DoesNotExist, ValueError, TypeError):
return Response(
{"detail": "systemVoiceId not found"},
status=status.HTTP_400_BAD_REQUEST,
)
else:
raw = data.get("ttsVoice")
if raw in (None, ""):
prefs.system_voice = None
else:
voice = get_voice(key=str(raw).strip())
if voice is None:
return Response(
{"detail": "invalid ttsVoice"},
status=status.HTTP_400_BAD_REQUEST,
)
prefs.system_voice = voice
update_fields.append("system_voice")
if "activeAgentId" in data:
from agency.secretary.resolution import is_agent_available_to_user
raw = data.get("activeAgentId")
if raw in (None, ""):
prefs.active_agent = None
else:
try:
agent = Agent.objects.get(pk=raw)
except (Agent.DoesNotExist, ValueError, TypeError):
return Response(
{"detail": "activeAgentId not found"},
status=status.HTTP_400_BAD_REQUEST,
)
if not is_agent_available_to_user(request.user, agent):
return Response(
{"detail": "activeAgentId not permitted"},
status=status.HTTP_400_BAD_REQUEST,
)
prefs.active_agent = agent
update_fields.append("active_agent")
if "activeThreadId" in data:
raw = data.get("activeThreadId")
if raw in (None, ""):
prefs.active_thread = None
else:
try:
thread = Thread.objects.get(pk=raw, user=request.user)
except (Thread.DoesNotExist, ValueError, TypeError):
return Response(
{"detail": "activeThreadId not found"},
status=status.HTTP_400_BAD_REQUEST,
)
prefs.active_thread = thread
update_fields.append("active_thread")
if update_fields:
prefs.save(update_fields=[*update_fields, "updated_at"])
return Response(prefs.to_api())
VIEW_SET_MIX = {
'list': mixins.ListModelMixin,
'create': mixins.CreateModelMixin,
'retrieve': mixins.RetrieveModelMixin,
'update': mixins.UpdateModelMixin,
'destroy': mixins.DestroyModelMixin,
}
[docs]
def view_set(cls, edit=None, actions=None, serializer_class=None):
"""Get the default model view set.
Args:
cls: Model class to create viewset for
edit: If False, only list/retrieve. If True, full CRUD. If None, uses actions.
actions: List of action names to include (e.g., ['list', 'create'])
serializer_class: Optional custom serializer class. If None, uses default serializer.
Returns:
ViewSet class for the given model
"""
if actions is None:
actions = []
if edit is False:
actions = ['list', 'retrieve']
if edit is True:
actions = ['list', 'retrieve', 'create', 'update', 'destroy']
summaries = {}
view_set = [viewsets.GenericViewSet]
for mixin in actions:
if isinstance(mixin, str):
summaries[mixin] = extend_schema(summary=f"{mixin} {cls.__name__}")
mixin = VIEW_SET_MIX[mixin]
view_set.insert(0, mixin)
schema_decorator = extend_schema_view(**summaries)
# Determinar el serializer a usar
final_serializer_class = serializer_class if serializer_class is not None else serializer(cls)
@schema_decorator
class ViewSet(*view_set):
serializer_class = final_serializer_class
permission_classes = [permissions.IsAuthenticatedOrReadOnly]
def get_queryset(self):
return cls.objects.all()
def perform_create(self, serializer):
serializer.save()
ViewSet.__name__ = cls.__name__ + "ViewSet"
return ViewSet
[docs]
@extend_schema(tags=['agent'])
@api_view(['POST'])
@permission_classes([permissions.IsAuthenticated])
def agent_test(request, agent_id):
"""Test an agent with a message.
Args:
request: HTTP request object
agent_id: UUID of the agent to test
Returns:
Response with agent test result or error message
"""
message = request.data.get('message') or request.query_params.get('message')
if not message:
return Response(
{'detail': 'Message parameter is required'},
status=status.HTTP_400_BAD_REQUEST
)
try:
agent = Agent.objects.get(id=agent_id)
except Agent.DoesNotExist:
return Response(
{'detail': 'Agent not found'},
status=status.HTTP_404_NOT_FOUND
)
try:
result = agent.test(message)
from agency.utils import serializable
serialized_result = serializable(result)
return Response(serialized_result, status=status.HTTP_200_OK)
except Exception as e:
return Response(
{'detail': str(e)},
status=status.HTTP_500_INTERNAL_SERVER_ERROR
)
[docs]
@extend_schema(tags=['mission'])
@api_view(['POST'])
@permission_classes([permissions.IsAuthenticated])
def mission_dispatch(request, mission_id):
"""Dispatch a mission with an agent.
Args:
request: HTTP request object with agent_id in body
mission_id: UUID of the mission to dispatch
Returns:
Response with execution_id, thread_id, and task_id
"""
agent_id = request.data.get('agent_id')
if not agent_id:
from agency.secretary.resolution import get_active_agent
agent = get_active_agent(request.user)
if agent is None:
return Response(
{'detail': 'No agent available; run boot or create Veronica first'},
status=status.HTTP_503_SERVICE_UNAVAILABLE,
)
else:
from agency.secretary.resolution import is_agent_available_to_user
try:
agent = Agent.objects.get(id=agent_id)
except Agent.DoesNotExist:
return Response(
{'detail': 'Agent not found'},
status=status.HTTP_404_NOT_FOUND,
)
if not is_agent_available_to_user(request.user, agent):
return Response(
{'detail': 'Agent not permitted'},
status=status.HTTP_400_BAD_REQUEST,
)
try:
mission = Mission.objects.get(id=mission_id)
except Mission.DoesNotExist:
return Response(
{'detail': 'Mission not found'},
status=status.HTTP_404_NOT_FOUND,
)
try:
from agency.secretary.dispatch import dispatch_mission_for_user
payload = dispatch_mission_for_user(request.user, mission, agent=agent)
return Response(payload, status=status.HTTP_200_OK)
except ValueError as e:
return Response(
{'detail': str(e)},
status=status.HTTP_400_BAD_REQUEST,
)
except Exception as e:
return Response(
{'detail': str(e)},
status=status.HTTP_500_INTERNAL_SERVER_ERROR
)
[docs]
def interruption_view_set():
"""Pending HITL list with agent display name."""
from agency.models import Interruption
class InterruptionSerializer(serializers.ModelSerializer):
agent_name = serializers.CharField(source='agent.name', read_only=True)
thread_id = serializers.SerializerMethodField()
class Meta:
model = Interruption
fields = (
'id',
'status',
'question',
'priority',
'agent',
'agent_name',
'user',
'thread_id',
'created_at',
'updated_at',
)
read_only_fields = (
'id',
'agent',
'agent_name',
'user',
'thread_id',
'created_at',
'updated_at',
)
def get_thread_id(self, obj):
try:
return str(obj.checkpoint.run.thread_id)
except Exception:
return None
base = view_set(
Interruption,
actions=['list', 'retrieve', 'update', 'destroy'],
serializer_class=InterruptionSerializer,
)
class InterruptionViewSet(base):
permission_classes = [permissions.IsAuthenticated]
def get_queryset(self):
qs = filter_by_owner(
Interruption.objects.select_related(
'agent',
'user',
'checkpoint',
'checkpoint__run',
'checkpoint__run__thread',
),
self.request.user,
)
status = self.request.query_params.get('status')
if status:
qs = qs.filter(status=status)
return qs.order_by('-created_at')
InterruptionViewSet.__name__ = 'InterruptionViewSet'
return InterruptionViewSet
[docs]
def photo_serializer():
"""Photo serializer: multipart file + catalog card fields."""
from agency.models import Photo
from agency.photos import photo_catalog_card
class PhotoSerializer(serializers.ModelSerializer):
file = serializers.ImageField(required=False)
class Meta:
model = Photo
fields = ("id", "title", "origin", "source", "file", "created_at")
read_only_fields = ("id", "created_at")
def to_representation(self, instance):
data = photo_catalog_card(instance)
request = self.context.get("request")
url = data.get("url")
if url and request is not None:
data["url"] = request.build_absolute_uri(url)
return data
PhotoSerializer.__name__ = "PhotoSerializer"
return PhotoSerializer
[docs]
def photo_view_set():
"""User-scoped photo library (uploads + generated)."""
from rest_framework.parsers import FormParser, JSONParser, MultiPartParser
from agency.models import Photo
ser = photo_serializer()
base = view_set(Photo, True, serializer_class=ser)
class PhotoViewSet(base):
permission_classes = [permissions.IsAuthenticated]
parser_classes = [JSONParser, MultiPartParser, FormParser]
def get_queryset(self):
qs = Photo.objects.filter(user=self.request.user)
origin = self.request.query_params.get("origin")
if origin:
qs = qs.filter(origin=origin)
return qs.order_by("-created_at")
def perform_create(self, serializer):
origin = serializer.validated_data.get("origin") or Photo.Origin.UPLOAD
source = serializer.validated_data.get("source") or "explorer"
serializer.save(
user=self.request.user,
origin=origin,
source=source,
)
def perform_destroy(self, instance):
if instance.file:
instance.file.delete(save=False)
instance.delete()
PhotoViewSet.__name__ = "PhotoViewSet"
return PhotoViewSet
[docs]
@extend_schema(tags=['provider'])
@api_view(['POST'])
@permission_classes([permissions.IsAuthenticated])
def provider_sync(request, provider_id):
"""Sync models from a provider API into local ``Model`` rows."""
try:
provider = Provider.objects.get(id=provider_id)
except Provider.DoesNotExist:
return Response(
{'detail': 'Provider not found'},
status=status.HTTP_404_NOT_FOUND,
)
try:
result = provider.sync_models()
return Response(result, status=status.HTTP_200_OK)
except Exception as e:
return Response(
{'detail': str(e)},
status=status.HTTP_500_INTERNAL_SERVER_ERROR,
)
# ── Corpus / RAG ──────────────────────────────────────────────────────────────
[docs]
def workspace_serializer():
from agency.corpus.models import Workspace
class WorkspaceSerializer(serializers.ModelSerializer):
tenant_name = serializers.CharField(source="tenant.name", read_only=True)
document_count = serializers.SerializerMethodField()
class Meta:
model = Workspace
fields = (
"id",
"tenant",
"tenant_name",
"name",
"slug",
"description",
"default_embedding_model",
"default_chunk_size",
"default_chunk_overlap",
"document_count",
"created_at",
)
read_only_fields = ("id", "created_at", "document_count", "tenant_name")
def get_document_count(self, obj):
return obj.documents.exclude(status="DELETED").count()
WorkspaceSerializer.__name__ = "WorkspaceSerializer"
return WorkspaceSerializer
[docs]
def document_serializer():
from agency.corpus.models import Document, Workspace
class WorkspaceBriefSerializer(serializers.ModelSerializer):
class Meta:
model = Workspace
fields = ("id", "name", "slug")
class DocumentSerializer(serializers.ModelSerializer):
workspace = WorkspaceBriefSerializer(read_only=True)
workspace_id = serializers.PrimaryKeyRelatedField(
queryset=Workspace.objects.all(),
source="workspace",
write_only=False,
)
file_url = serializers.SerializerMethodField()
class Meta:
model = Document
fields = (
"id",
"workspace",
"workspace_id",
"title",
"file",
"file_url",
"source_url",
"summary",
"checksum_sha256",
"status",
"error_message",
"page_count",
"chunk_count",
"indexed_at",
"created_at",
"updated_at",
)
read_only_fields = (
"id",
"checksum_sha256",
"status",
"error_message",
"page_count",
"chunk_count",
"indexed_at",
"summary",
"created_at",
"updated_at",
"file_url",
)
extra_kwargs = {
"file": {"required": False, "allow_null": True},
"source_url": {"required": False, "allow_blank": True},
}
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
request = self.context.get("request")
if request is not None:
self.fields["workspace_id"].queryset = filter_by_tenant_membership(
Workspace.objects.all(),
request.user,
)
def get_file_url(self, obj):
if not obj.file:
return None
request = self.context.get("request")
url = obj.file.url
if request is not None:
return request.build_absolute_uri(url)
return url
DocumentSerializer.__name__ = "DocumentSerializer"
return DocumentSerializer
[docs]
def workspace_view_set():
from agency.corpus.models import Workspace
ser = workspace_serializer()
base = view_set(Workspace, True, serializer_class=ser)
class WorkspaceViewSet(base):
permission_classes = [permissions.IsAuthenticated]
def get_queryset(self):
return filter_by_tenant_membership(
Workspace.objects.select_related("tenant")
.prefetch_related("documents")
.order_by("tenant__name", "name"),
self.request.user,
)
WorkspaceViewSet.__name__ = "WorkspaceViewSet"
return WorkspaceViewSet
[docs]
def document_view_set():
from rest_framework.parsers import FormParser, JSONParser, MultiPartParser
from agency.corpus.models import Document
from agency.corpus.store import Collection
ser = document_serializer()
base = view_set(Document, True, serializer_class=ser)
class DocumentViewSet(base):
permission_classes = [permissions.IsAuthenticated]
parser_classes = [JSONParser, MultiPartParser, FormParser]
def get_queryset(self):
qs = filter_documents_by_tenant_membership(
Document.objects.select_related("workspace").exclude(status="DELETED"),
self.request.user,
)
workspace = self.request.query_params.get("workspace")
if workspace:
qs = qs.filter(workspace_id=workspace)
status_filter = self.request.query_params.get("status")
if status_filter:
qs = qs.filter(status=status_filter)
return qs.order_by("-created_at")
def perform_destroy(self, instance):
try:
Collection(instance.workspace).delete_document(instance)
except Exception:
pass
instance.delete()
DocumentViewSet.__name__ = "DocumentViewSet"
return DocumentViewSet
def _serialize_search_hit(doc, score=None):
meta = dict(getattr(doc, "metadata", None) or {})
if score is not None and "score" not in meta:
meta["score"] = score
return {
"text": getattr(doc, "page_content", None) or getattr(doc, "text", "") or "",
"score": float(score) if score is not None else meta.get("score"),
"metadata": meta,
}
[docs]
@extend_schema(tags=["document"])
@api_view(["POST"])
@permission_classes([permissions.IsAuthenticated])
def document_reindex(request, document_id):
"""Queue Celery ``index_document`` for an existing document."""
from agency.corpus.models import Document
from agency.corpus.tasks import index_document
document = filter_documents_by_tenant_membership(
Document.objects.select_related("workspace"),
request.user,
).filter(pk=document_id).first()
if document is None:
return Response({"detail": "Document not found"}, status=status.HTTP_404_NOT_FOUND)
if not document.file:
return Response(
{"detail": "Document has no file to index"},
status=status.HTTP_400_BAD_REQUEST,
)
document.status = Document.Status.UPLOADED
document.error_message = ""
document.save(update_fields=["status", "error_message", "updated_at"])
task = index_document.delay(str(document.id), 0)
return Response(
{"task_id": task.id, "document_id": str(document.id), "status": document.status},
status=status.HTTP_200_OK,
)
[docs]
@extend_schema(tags=["workspace"])
@api_view(["POST"])
@permission_classes([permissions.IsAuthenticated])
def workspace_search(request, workspace_id):
"""Similarity search against the workspace Chroma collection."""
from agency.corpus.models import Document as DocModel
from agency.corpus.models import Workspace
from agency.corpus.generation import Corpus
from agency.corpus.generation.query import Query
workspace = filter_by_tenant_membership(
Workspace.objects.all(),
request.user,
).filter(pk=workspace_id).first()
if workspace is None:
return Response({"detail": "Workspace not found"}, status=status.HTTP_404_NOT_FOUND)
query_text = (request.data.get("query") or "").strip()
if not query_text:
return Response({"detail": "query is required"}, status=status.HTTP_400_BAD_REQUEST)
k = int(request.data.get("k") or 10)
document_id = request.data.get("document_id")
try:
if document_id:
try:
doc = DocModel.objects.get(id=document_id, workspace=workspace)
except DocModel.DoesNotExist:
return Response(
{"detail": "Document not found in workspace"},
status=status.HTTP_404_NOT_FOUND,
)
docs = Query(str(workspace.id)).find(
query_text=query_text,
similarity_top_k=k,
document=doc,
)
hits = [_serialize_search_hit(d) for d in docs]
else:
results = Corpus(workspace).search(query_text)
hits = []
for item in results[:k]:
if isinstance(item, (list, tuple)) and len(item) == 2:
hits.append(_serialize_search_hit(item[0], item[1]))
else:
hits.append(_serialize_search_hit(item))
return Response(
{
"workspace_id": str(workspace.id),
"query": query_text,
"hits": hits,
}
)
except Exception as e:
return Response({"detail": str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
[docs]
@extend_schema(tags=["workspace"])
@api_view(["POST"])
@permission_classes([permissions.IsAuthenticated])
def workspace_ask(request, workspace_id):
"""RAG ask: retrieve + LLM answer with source nodes."""
from agency.corpus.models import Workspace
from agency.corpus.generation.query import Query
workspace = filter_by_tenant_membership(
Workspace.objects.all(),
request.user,
).filter(pk=workspace_id).first()
if workspace is None:
return Response({"detail": "Workspace not found"}, status=status.HTTP_404_NOT_FOUND)
query_text = (request.data.get("query") or "").strip()
if not query_text:
return Response({"detail": "query is required"}, status=status.HTTP_400_BAD_REQUEST)
k = int(request.data.get("k") or 10)
try:
engine = Query(str(workspace.id)).query_text(query_text, similarity_top_k=k)
response = engine.query(query_text)
hits = [
{
"text": node.text,
"score": node.score,
"metadata": node.metadata or {},
}
for node in response.source_nodes
]
return Response(
{
"workspace_id": str(workspace.id),
"query": query_text,
"answer": str(response),
"hits": hits,
}
)
except Exception as e:
return Response({"detail": str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
# ── Staff API ──────────────────────────────────────────────────────────────────
[docs]
def user_serializer():
"""User serializer for staff explorer (no password hash exposure)."""
class GroupBriefSerializer(serializers.ModelSerializer):
class Meta:
model = Group
fields = ("id", "name")
class UserSerializer(serializers.ModelSerializer):
groups = GroupBriefSerializer(many=True, read_only=True)
group_ids = serializers.PrimaryKeyRelatedField(
many=True,
queryset=Group.objects.all(),
source="groups",
write_only=True,
required=False,
)
class Meta:
model = User
fields = (
"id",
"username",
"email",
"first_name",
"last_name",
"is_active",
"is_staff",
"is_superuser",
"groups",
"group_ids",
"date_joined",
"last_login",
)
read_only_fields = ("id", "date_joined", "last_login")
UserSerializer.__name__ = "UserSerializer"
return UserSerializer
[docs]
def user_view_set():
"""Staff-only User list/retrieve/update viewset."""
ser = user_serializer()
base = view_set(User, actions=["list", "retrieve", "update"], serializer_class=ser)
class UserViewSet(base):
permission_classes = [permissions.IsAdminUser]
def get_queryset(self):
return User.objects.prefetch_related("groups").order_by("username")
UserViewSet.__name__ = "UserViewSet"
return UserViewSet
[docs]
def group_serializer():
"""Group serializer with permission ids."""
class GroupSerializer(serializers.ModelSerializer):
permission_ids = serializers.PrimaryKeyRelatedField(
many=True,
queryset=Permission.objects.all(),
source="permissions",
required=False,
)
class Meta:
model = Group
fields = ("id", "name", "permission_ids")
GroupSerializer.__name__ = "GroupSerializer"
return GroupSerializer
[docs]
def group_view_set():
"""Staff-only Group list/create/update viewset."""
ser = group_serializer()
base = view_set(Group, actions=["list", "retrieve", "create", "update"], serializer_class=ser)
class GroupViewSet(base):
permission_classes = [permissions.IsAdminUser]
def get_queryset(self):
return Group.objects.prefetch_related("permissions").order_by("name")
GroupViewSet.__name__ = "GroupViewSet"
return GroupViewSet
[docs]
@extend_schema(tags=["staff"])
class UserPrefsStaff(views.APIView):
"""Staff-only access to any user's preferences."""
permission_classes = [permissions.IsAdminUser]
schema = AutoSchema()
[docs]
@extend_schema(summary="Get user preferences (staff)")
def get(self, request, user_id):
try:
target_user = User.objects.get(pk=user_id)
except (User.DoesNotExist, ValueError, TypeError):
return Response(
{"detail": "User not found"},
status=status.HTTP_404_NOT_FOUND,
)
prefs = UserPreferences.for_user(target_user)
return Response(prefs.to_api())
[docs]
@extend_schema(summary="Update user preferences (staff)")
def patch(self, request, user_id):
try:
target_user = User.objects.get(pk=user_id)
except (User.DoesNotExist, ValueError, TypeError):
return Response(
{"detail": "User not found"},
status=status.HTTP_404_NOT_FOUND,
)
prefs = UserPreferences.for_user(target_user)
data = request.data if isinstance(request.data, dict) else {}
update_fields: list[str] = []
if "workspaceId" in data:
raw = data.get("workspaceId")
if raw in (None, ""):
prefs.active_workspace = None
else:
from agency.corpus.models import Workspace
try:
prefs.active_workspace = Workspace.objects.get(pk=raw)
except (Workspace.DoesNotExist, ValueError, TypeError):
return Response(
{"detail": "workspaceId not found"},
status=status.HTTP_400_BAD_REQUEST,
)
update_fields.append("active_workspace")
if "desktop" in data:
desktop = data.get("desktop")
if desktop is None:
prefs.desktop = {}
elif isinstance(desktop, dict):
from agency.media import preserve_now_playing
prefs.desktop = preserve_now_playing(prefs.desktop, desktop)
else:
return Response(
{"detail": "desktop must be an object"},
status=status.HTTP_400_BAD_REQUEST,
)
update_fields.append("desktop")
if "language" in data:
from agency.ui.language import normalize_language
raw = data.get("language")
if raw in (None, ""):
prefs.language = ""
else:
prefs.language = normalize_language(str(raw))
update_fields.append("language")
if "timezone" in data:
from agency.ui.timezone import normalize_timezone
raw = data.get("timezone")
if raw in (None, ""):
prefs.timezone = ""
else:
try:
prefs.timezone = normalize_timezone(str(raw))
except ValueError:
return Response(
{"detail": "invalid timezone"},
status=status.HTTP_400_BAD_REQUEST,
)
update_fields.append("timezone")
if "systemVoiceId" in data or "ttsVoice" in data:
from agency.voices import get_voice
if "systemVoiceId" in data:
raw = data.get("systemVoiceId")
if raw in (None, ""):
prefs.system_voice = None
else:
try:
prefs.system_voice = Voice.objects.get(pk=raw)
except (Voice.DoesNotExist, ValueError, TypeError):
return Response(
{"detail": "systemVoiceId not found"},
status=status.HTTP_400_BAD_REQUEST,
)
else:
raw = data.get("ttsVoice")
if raw in (None, ""):
prefs.system_voice = None
else:
voice = get_voice(key=str(raw).strip())
if voice is None:
return Response(
{"detail": "invalid ttsVoice"},
status=status.HTTP_400_BAD_REQUEST,
)
prefs.system_voice = voice
update_fields.append("system_voice")
if "activeAgentId" in data:
raw = data.get("activeAgentId")
if raw in (None, ""):
prefs.active_agent = None
else:
try:
agent = Agent.objects.get(pk=raw)
except (Agent.DoesNotExist, ValueError, TypeError):
return Response(
{"detail": "activeAgentId not found"},
status=status.HTTP_400_BAD_REQUEST,
)
prefs.active_agent = agent
update_fields.append("active_agent")
if "activeThreadId" in data:
raw = data.get("activeThreadId")
if raw in (None, ""):
prefs.active_thread = None
else:
try:
thread = Thread.objects.get(pk=raw, user=target_user)
except (Thread.DoesNotExist, ValueError, TypeError):
return Response(
{"detail": "activeThreadId not found"},
status=status.HTTP_400_BAD_REQUEST,
)
prefs.active_thread = thread
update_fields.append("active_thread")
if update_fields:
prefs.save(update_fields=[*update_fields, "updated_at"])
return Response(prefs.to_api())