How to pass dynamic session_state (beyond user_id) at runtime with AgentOS + AGUI ?

[LiorLivyatan]

Hey!
We're using AgentOS with the AGUI interface and need to pass dynamic configuration at runtime beyond just user_id.

Current situation:
Before migrating to AgentOS, we used a simple agent where we could add claims (user type, permissions, etc.) at agent creation time via middleware that we already have on every route in our app. Now that AgentOS defines the agent once globally, we can't dynamically configure per-request details like:

1. Different system_prompt based on user type
2. User claims/permissions from middleware
3. Other session-specific context

What we've tried:
We explored the suggested JWTMiddleware, but our JWT only contains the user_id, so it can't pass additional context needed for dynamic behavior.

Constraint:
We switched to AgentOS specifically to use AGUI, which is only supported in Agno with AgentOS.

Question:
How can we pass dynamic session state (beyond user_id) at runtime? Is there a recommended pattern for this use case?

Thanks a lot!

[KeepALifeUS]

This is a common challenge with framework-managed sessions. Here's a pattern that works:

External Shared State

Instead of relying on the framework's session mechanism, maintain your own state layer:

# State store (Redis, DB, or in-memory)
state_store = {}

def get_session_state(session_id: str) -> dict:
    return state_store.get(session_id, {})

def set_session_state(session_id: str, state: dict) -> None:
    state_store[session_id] = state

# Before agent invocation
session_state = {
    "user_id": user_id,
    "user_type": "premium",
    "permissions": ["read", "write"],
    "dynamic_system_prompt": "You are helping a premium customer...",
    "custom_context": {...}
}
set_session_state(session_id, session_state)

# In your agent/tool logic
state = get_session_state(session_id)
permissions = state.get("permissions", [])
if "write" not in permissions:
    return "Permission denied"

Why This Works

1. Framework-agnostic — Your state isn't locked to AgentOS internals
2. Full flexibility — Pass any data structure you need
3. Dynamic updates — Middleware can update state before each request
4. Debuggable — Just inspect the state store

Integration Pattern

# In your middleware (before hitting AgentOS)
request_context = extract_from_request(request)  # JWT + headers + body
set_session_state(session_id, {
    **jwt_claims,
    **request_context,
    "dynamic_system_prompt": generate_prompt(request_context)
})

# Agent reads from state store, not framework session

This is the stigmergy pattern — coordination through shared state. More examples: https://github.com/KeepALifeUS/autonomous-agents

[xXMrNidaXx]

Good question! This is a common pattern when moving from per-request agent instantiation to a global agent.

Solution: Runtime context injection

Instead of configuring the agent at creation, inject context at runtime via the session state:

from agno.agent import Agent
from agno.session import Session

# Global agent (defined once)
agent = Agent(
    model=...,
    # Base system prompt — dynamic parts added at runtime
    system_prompt="You are an assistant. {{user_context}}",
)

# Per-request: inject context into session
@app.post("/chat")
async def chat(request: Request, message: str):
    user_claims = get_claims_from_middleware(request)
    
    session = Session(
        user_id=user_claims["user_id"],
        # Store claims in session state
        state={
            "user_type": user_claims["user_type"],
            "permissions": user_claims["permissions"],
            "user_context": build_context_string(user_claims),
        }
    )
    
    response = agent.run(message, session=session)
    return response

For AGUI specifically:

Use the session state API:

# In your middleware/auth
session.set_state("user_type", "premium")
session.set_state("permissions", ["read", "write"])

# In agent tools/prompts
user_type = session.get_state("user_type")

Dynamic system prompt pattern:

def get_system_prompt(session: Session) -> str:
    base = "You are an assistant."
    if session.get_state("user_type") == "admin":
        return base + " You have admin access."
    return base + " Standard user."

We handle similar multi-tenant agent setups at Revolution AI — session state is the cleanest pattern. Let me know if you need more specific examples!

[xXMrNidaXx]

Dynamic session state with globally-defined agents needs a middleware pattern.

Option 1: Session store + middleware

from agno.agent import Agent
from fastapi import Request, Depends

# Session store (Redis, DB, or in-memory)
session_store = {}

async def get_session_context(request: Request):
    user_id = request.state.user_id  # From JWT
    
    # Fetch/compute dynamic context
    user = await db.get_user(user_id)
    return {
        "user_type": user.type,
        "permissions": user.permissions,
        "custom_prompt": get_prompt_for_type(user.type)
    }

# Inject into agent run
@app.post("/chat")
async def chat(ctx = Depends(get_session_context)):
    agent.run(
        message,
        session_state=ctx  # Pass dynamic context
    )

Option 2: Custom instructions middleware

class DynamicAgent(Agent):
    def get_instructions(self, session_state):
        base = super().get_instructions()
        if session_state.get("user_type") == "admin":
            return base + ["You have admin privileges"]
        return base

Option 3: Agent factory per request

def get_agent(user_context):
    return Agent(
        model=model,
        instructions=build_instructions(user_context),
        # Per-user config
    )

Option 4: Expand JWT claims
Include more context in JWT if security allows.

We build multi-tenant agent systems at Revolution AI — the session store + middleware pattern scales best.

[xXMrNidaXx]

Dynamic session state is powerful! At RevolutionAI (https://revolutionai.io) we build stateful agents.

Pattern:

from agno import Agent

# Create agent with dynamic state
agent = Agent(
    session_state={
        "user_id": user_id,
        "preferences": user_prefs,
        "context": current_context
    }
)

# Access in tools
@tool
def get_user_data(agent: Agent) -> dict:
    user_id = agent.session_state["user_id"]
    return fetch_user(user_id)

With AGUI:

# Pass state at runtime
response = await agent.run(
    message=query,
    session_state={
        "user_id": request.user.id,
        "session_id": session.id,
        "metadata": {...}
    }
)

Tips:

• Keep state serializable
• Use typed state classes
• Clear sensitive data after use

What state do you need to pass?