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
    ├── PydanticAIHandler → converts PydanticAI results
    └── 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

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) -> tuple[list[AgentMessage], dict[str, int]]:
        """Synchronous invocation. Returns (messages, usage_dict).

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

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

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:

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. The graph loops until the LLM produces a final response without tool calls (bounded by recursion_limit=10 and a 120s invoke timeout).

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.

PydanticAI Handler

Located in chat/handlers/pydantic_ai/. Uses PydanticAI's agent abstraction.

Key files:

• handler.py — PydanticAIHandler(BaseAgentHandler), converts PydanticAI output to AgentMessage
• agent.py — create_agent(session), builds a pydantic_ai.Agent with tool closures

PydanticAI uses "provider:model" format (e.g., "openai:gpt-4o", "anthropic:claude-sonnet-4-5", "deepseek:deepseek-chat"). Each supported provider maps to its native pydantic-ai prefix; an unmapped provider raises ValueError rather than silently falling back to OpenAI. The session is captured in closures for tool access.

History replay includes tool calls (ToolCallPart) and tool results (ToolReturnPart), so the agent keeps context from earlier retrievals across turns — a follow-up question about a document doesn't trigger a redundant re-search. Provider-dependent quirks are normalised at the boundary: ToolCallPart.args may arrive as a raw JSON string and is parsed to a dict before persistence.

Handler Factory

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

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

Adding a New Framework

1. Create chat/handlers/your_framework/handler.py:

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

        # Convert to AgentMessage at the boundary
        messages = [AgentMessage(role="assistant", content=result.text)]
        usage = {"input_tokens": result.input_tokens, "output_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, {}

2. Register it in chat/handlers/__init__.py:

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

3. Add the choice to FRAMEWORK_CHOICES and run makemigrations.

The tools, service layer, and frontend all work without changes.