Admin Chat Agent

The admin chat is a natural-language interface embedded directly in the Django admin dashboard at /admin/. Superusers can type questions and the MCP agent answers them by calling Django tools — querying models, introspecting the app, running management commands — and streaming results back to the browser in real time.

No external server is needed. Everything runs inside the same Django process.

Endpoints

POST /cfg/mcp/agent/stream/

The SSE streaming endpoint used by the admin chat UI.

Auth: Django session cookie + CSRF token. The view requires is_authenticated; the admin template gates the UI to {% if user.is_superuser %}.

Request body:

{
  "message": "How many users signed up this week?",
  "session_id": "admin-chat",
  "model": "openai/gpt-4o-mini"
}

session_id and model are optional. session_id defaults to "agent-session".

Response: Content-Type: text/event-stream

event: text
data: {"content": "There are 42 users who signed up this week."}

event: done
data: {}

The agent runs its tool calls internally and streams back only the final answer. Intermediate tool activity is not surfaced as SSE events (see SSE event types below).

POST /cfg/mcp/agent/

Synchronous variant. Uses X-MCP-Access-Key header auth (not session). Blocks until the agent finishes all tool calls, then returns one JSON response. Useful for programmatic clients; not used by the admin UI.

SSE event types

The stream carries only the final answer. Tool calls run inside the agent and are not emitted as tool_start / tool_result events — the agent runs to completion server-side, then the resulting text is streamed. (An earlier version surfaced per-tool events; that was removed because OpenRouter streaming could complete without text deltas and trigger a duplicate provider request.)

Agentic loop

agents/api/views.py:MCPAgentStreamView runs the agent synchronously via agent_runner.run(...) and turns the result into the SSE stream:

user_message + Redis history → AgentContext (with user=request.user, request=request)
tools_def = OpenAI-compatible function definitions from the registry

result = agent_runner.run(message, context, model=model)   # runs all tool calls internally
yield {"event": "text", "content": result.text}
yield {"event": "done"}
# on failure: yield {"event": "error", "message": ...}; yield {"event": "done"}

AgentContext exposes user and request to tools, so a custom tool can act with the authenticated request in scope. Django’s StreamingHttpResponse sends each yielded chunk immediately because Gunicorn/uvicorn flushes after each yield.

Session history

History is persisted in Redis (Django’s default cache), keyed as mcp_chat:{user_id}:{session_id}. On each request MCPAgentStreamView:

1. Loads prior history via RedisMCPChatStore and seeds AgentContext with it.
2. Appends the user message.
3. After the stream ends, appends the assistant reply.

So a conversation continues across requests as long as the same session_id is sent. See Redis & Caching for the store API and key format.

Browser-side implementation

The chat UI uses fetch with ReadableStream rather than EventSource because the request is a POST with a JSON body:

const resp = await fetch('/cfg/mcp/agent/stream/', {
    method: 'POST',
    headers: {
        'Content-Type': 'application/json',
        'X-CSRFToken': getCsrf(),   // reads csrftoken cookie
    },
    body: JSON.stringify({ message, session_id: 'admin-chat' }),
});
 
const reader = resp.body.getReader();
// reads chunks, splits on '\n\n', parses 'event:' and 'data:' lines

The CSRF token is read from the csrftoken cookie set automatically by Django’s CSRF middleware for authenticated sessions.

How the view wires Redis history

from django_cfg.modules.django_mcp.chat.store import RedisMCPChatStore
 
store = RedisMCPChatStore(user_id=request.user.pk, session_id=session_id)
history = store.get_history()   # list of {"role": ..., "content": ...}
 
context = AgentContext(
    tools=tools, session_key=session_id, config=mcp_config,
    user=request.user, request=request,
)
for msg in history:
    context.add_message(msg["role"], msg["content"])
 
result = agent_runner.run(message, context, model=model)   # tools run internally
 
# Persist both turns
store.append_message("user", message)
if result.text:
    store.append_message("assistant", result.text)
 
def event_stream():
    yield _sse_line({"event": "text", "content": result.text})
    yield _sse_line({"event": "done"})