Explain Session and State Management in Google ADK with state prefix scoping (app, user, temp).

Session & State Management in Google ADK

Google ADK provides a comprehensive session and state management system with prefix-based scoping to control data visibility and lifecycle.

Session Lifecycle

State Prefix Scoping

app:

app:model_version

user:

user:preferences

temp:

temp:intermediate_result

query

docs

pythonCopy
# Setting state in a tool
def my_tool(query: str, tool_context) -> str:
    """Process a query with state management."""
    # App-wide state
    tool_context.state["app:total_queries"] = (
        tool_context.state.get("app:total_queries", 0) + 1
    )

    # User-scoped state
    tool_context.state["user:last_query"] = query
    tool_context.state["user:query_count"] = (
        tool_context.state.get("user:query_count", 0) + 1
    )

    # Session-scoped state
    tool_context.state["current_query"] = query

    # Temp state (only this invocation)
    tool_context.state["temp:processing"] = True

    return f"Processed: {query}"

Session Services

InMemorySessionService

DatabaseSessionService

VertexAiSessionService

pythonCopy
from google.adk.sessions import InMemorySessionService, DatabaseSessionService
from google.adk.runners import Runner

# Development
dev_sessions = InMemorySessionService()

# Production
prod_sessions = DatabaseSessionService(
    db_url="postgresql://user:pass@host:5432/dbname"
)

runner = Runner(
    agent=root_agent,
    app_name="my_app",
    session_service=prod_sessions,
)

Dynamic Instructions with State

Agent instructions can use

{variable}

pythonCopy
agent = Agent(
    name="personalized_agent",
    model="gemini-2.5-flash",
    instruction="""You are a helpful assistant for {user:name}.
    Their preferred language is {user:language}.
    Current query count: {user:query_count}.
    Model version: {app:model_version}.""",
)

Artifacts (Persistent Outputs)

pythonCopy
from google.genai import types

async def generate_report(query: str, tool_context) -> str:
    """Generate and save a report."""
    report_content = f"Report for: {query}\n..."

    artifact = types.Part.from_data(
        data=report_content.encode(),
        mime_type="text/plain"
    )
    version = await tool_context.save_artifact("report.txt", artifact)

    return f"Report saved as report.txt (version {version})"

# Load artifact later
artifact = await tool_context.load_artifact("report.txt")

Session Rewind

Roll back a session to before a specific invocation:

pythonCopy
# Rewind to before the last invocation
session = runner.session_service.get_session(
    app_name="my_app",
    user_id="user-123",
    session_id="session-abc"
)

# Rewind removes the last invocation's events and state changes
runner.session_service.rewind_session(session, invocation_id="inv-xyz")

Memory Service (Cross-Session)

pythonCopy
from google.adk.memory import InMemoryMemoryService

memory = InMemoryMemoryService()

runner = Runner(
    agent=root_agent,
    app_name="my_app",
    session_service=InMemorySessionService(),
    memory_service=memory,   # enables cross-session recall
)

Learn more at Sessions and Runtime.