Agent Handlers

HyperSaaS uses a framework-agnostic agent pattern. The service layer never imports any AI framework — each handler converts its native output to a common AgentMessage at the boundary.

Architecture

services.py (zero framework imports)
    │
    ▼
AgentMessage dataclass ← common boundary type
    │
    ├── LangGraphHandler → converts LangChain messages
    └── YourCustomHandler → convert anything

AgentMessage

The framework-agnostic message type defined in chat/handlers/base.py:

@dataclass
class AgentMessage:
    role: str               # "assistant" | "tool"
    content: str
    tool_calls: list[dict] | None = None  # [{"id", "name", "args"}]
    tool_call_id: str | None = None

StreamEvent

What a handler yields while it streams, also in chat/handlers/base.py:

@dataclass
class StreamEvent:
    kind: str                          # "text", "tool" or "final"
    text: str = ""                     # the next chunk of reply text ("text")
    tool: dict | None = None           # {"id", "name", "status": "started" | "finished"} ("tool")
    response: Any = None               # what invoke() would return first ("final")
    usage: dict[str, int] | None = None  # token usage ("final")

The view relays text and tool events to the client as server-sent events, then saves the reply from the final one. See API Endpoints.

BaseAgentHandler

The abstract base class all handlers implement:

class BaseAgentHandler(ABC):
    def __init__(self, session: ChatSession):
        self.session = session

    @abstractmethod
    def invoke(
        self, user_message: str, images: list | None = None, reply_to=None
    ) -> tuple[list[AgentMessage], dict[str, int]]:
        """Synchronous invocation. Returns (messages, usage) with usage keys
        prompt_tokens, completion_tokens and total_tokens.

        Raises AgentInvocationError on timeout or failure.
        """
        ...

    def stream(
        self, user_message: str, images: list | None = None, reply_to=None
    ) -> Iterator[StreamEvent]:
        """Yield "text" and "tool" events as they happen, then one "final" event.

        Optional: callers catch NotImplementedError and fall back to invoke().
        """
        raise NotImplementedError

    @abstractmethod
    async def astream(self, user_message: str) -> AsyncGenerator:
        """Async streaming. Yields (chunk, usage) tuples."""
        ...

images are the images attached to this message, for models that accept them. reply_to is the saved user message being answered: history is taken from before it, which is how regenerating or editing an earlier message works. stream() is a plain generator because the backend runs on WSGI, where Django would buffer an async iterator in full.

Failure contract

Handlers raise AgentInvocationError (also in handlers/base.py) on timeout or any failure — never an empty result. This lets callers distinguish "the agent failed" (surface an error to the user) from "the agent legitimately produced nothing." The DRF views convert the exception into a user-visible validation error; the Celery task retries once before reporting failure.

Shared Modules

Framework handlers stay thin because the cross-cutting concerns live in shared modules:

Module	Responsibility
`chat/handlers/prompts.py`	System-prompt assembly for all frameworks: knowledge-base policy, citation rules, the `map_locations` JSON protocol, and error-handling instructions. It auto-detects whether the session has knowledge bases attached and tells the model definitively — KBs attached → an unconditional "search the knowledge base first" instruction; none → "don't call the search tool."
`chat/handlers/history.py`	`build_history(session, reply_to, prompt)` returns framework-neutral message dicts (`role`, `content`, `tool_calls`, `tool_call_id`) for as much recent history as fits the token budget, plus a summary of what was left out. Each handler converts these to its native message types, so history behavior can't drift between frameworks.

Because both frameworks share the prompt builder, sessions behave the same regardless of agent_framework — same retrieval policy, same citation rules, same map protocol.

LangGraph Handler

Located in chat/handlers/langgraph/. Uses LangGraph's graph-based execution with tool calling.

Key files:

handler.py — LangGraphHandler(BaseAgentHandler), converts output to AgentMessage
graph.py — Defines the LangGraph state machine (llm node → tool node → loop)
nodes.py — LLM call node, tool execution node, routing logic
state.py — Graph state definition
tools.py — Wraps plain Python tools with LangChain @tool decorators

Flow:

User message → LLM node → should_continue? → tool node → LLM node → ... → END

The LLM decides when to call tools, except that with knowledge bases attached, the first step for each question is a knowledge-base search. The graph loops until the LLM produces a final response without tool calls (bounded by recursion_limit=10 and a 120s timeout). Streaming relays the LLM's tokens and each tool call as it starts and finishes.

The handler prebuilds the tool-bound LLM client and the system prompt once per invoke and passes them through the graph config — nodes never reconstruct them on each loop iteration.

Handler Factory

# chat/handlers/__init__.py
FRAMEWORK_CHOICES = [
    ("none", "None"),
    ("langgraph", "LangGraph"),
]

def get_agent_handler(session) -> BaseAgentHandler | None:
    framework = getattr(session, "agent_framework", "none")
    if framework == "langgraph":
        return LangGraphHandler(session)
    return None

Adding a New Framework

Create chat/handlers/your_framework/handler.py:

class YourHandler(BaseAgentHandler):
    def invoke(self, user_message: str, images=None, reply_to=None):
        # Call your framework
        result = your_framework.run(user_message)

        # Convert to AgentMessage at the boundary
        messages = [AgentMessage(role="assistant", content=result.text)]
        usage = {
            "prompt_tokens": result.input_tokens,
            "completion_tokens": result.output_tokens,
            "total_tokens": result.input_tokens + result.output_tokens,
        }
        return messages, usage

    async def astream(self, user_message: str):
        async for chunk in your_framework.stream(user_message):
            yield chunk.text, {}

elif framework == "your_framework":
    return YourHandler(session)

Add the choice to FRAMEWORK_CHOICES and run makemigrations.

The tools, service layer, and frontend all work without changes. Implement stream() too for streamed replies; without it, the streaming endpoints return the reply as JSON once it's done. If your framework calls the paid tools, record them with record_tool_call as the LangGraph handler does.

On this page