Building Persistent Sessions with Google ADK: A Comprehensive Guide

Imagine having a conversation with someone who forgets everything you told them the moment you say goodbye.

Every time you meet, you start from scratch, repeating your name, preferences, and context all over again. Frustrating, right?

This is exactly what happens with stateless AI agents—they suffer from digital amnesia, losing all context between interactions.

But what if your AI agent could remember?

What if it could recall your preferences, build on previous conversations, and provide truly personalized experiences?

This is where persistent sessions transform everything.

In this comprehensive guide, we'll dive deep into building stateful AI agents using Google's Agent Development Kit (ADK), turning forgetful bots into intelligent assistants with real memory.

Understanding Persistent Sessions: The Foundation of Stateful AI

Persistent sessions represent the backbone of any meaningful AI agent interaction.

At their core, persistent sessions enable agents to maintain state across multiple conversations and interactions.

Unlike stateless systems that treat each request in isolation, persistent sessions create continuity and context. This continuity transforms simple question-answering bots into intelligent assistants that understand history and preferences.

The concept mirrors human conversation—we naturally remember past discussions and use that context to inform future interactions. When you tell a friend your favorite food once, they remember it for future conversations.

Persistent sessions give AI agents this same capability through structured state management. The state persists beyond the lifecycle of a single request-response cycle, living in a database that survives application restarts.

Google ADK implements persistent sessions through its DatabaseSessionService component. This service handles all the complexity of state serialization, storage, and retrieval.

Developers simply define what state to track, and the framework manages persistence automatically. The result is agents that feel more natural, contextual, and useful to end users.

Architecture Overview: Components of a Persistent Agent

The persistent session agent architecture consists of several interconnected components working in harmony.

Understanding this architecture is crucial before implementing your own agent. Each component has a specific responsibility, following separation of concerns principles. Let's break down the key players in this system.

DatabaseSessionService acts as the persistence layer, managing all database operations. It handles session creation, retrieval, updates, and listing across different users and applications. The service abstracts away SQL complexity, providing a clean interface for state management. In our example, it uses SQLite, but production systems might use PostgreSQL or other robust databases.

Agent represents the AI brain of your system, powered by Google's Gemini models. The agent receives instructions that define its personality and capabilities. It has access to custom tools that extend its functionality beyond text generation. Think of the agent as the decision-maker that orchestrates tool calls and generates responses.

Runner orchestrates the interaction between the agent and the session service. It manages the request-response cycle, handling message routing and state updates. The runner ensures that state changes persist correctly after each interaction. This component bridges the gap between your application logic and the AI agent.

Custom Tools extend the agent's capabilities with domain-specific functionality. These functions receive a ToolContext parameter that provides access to session state. Tools can read from state, modify state, or perform external operations. In our example, get_user_state and update_preference demonstrate state management patterns.

The data flow follows this pattern: user sends message → runner receives it → agent processes with tools → state updates → response returns. This unidirectional flow ensures consistency and makes debugging straightforward. Each component has clear inputs and outputs, promoting maintainability.

Defining Your Session State Structure

Session state structure determines what information your agent remembers across conversations. Thoughtful state design directly impacts the quality and usefulness of persistent sessions. You want to balance between storing enough context and avoiding unnecessary data bloat. The structure should align with your agent's purpose and user needs.

In our example, we initialize state with a simple preferences dictionary:

def initial_state():
    """Initialize the session state with default preferences."""
    return {
        "preferences": {
            "favorite_color": "blue",
            "favorite_food": "pizza"
        }
    }

This structure creates a nested dictionary where preferences live under a dedicated key. The nesting allows for future expansion—you could add other top-level keys like conversation_history or user_profile. Starting with sensible defaults ensures the agent functions properly on first use. Users can then modify these defaults through natural language interactions.

State structure should follow these principles: use descriptive keys, maintain consistency, enable extensibility, and serialize cleanly. Descriptive keys make your code self-documenting and easier to maintain. Consistency across sessions prevents confusion and bugs when accessing state. Extensibility allows adding new state fields without breaking existing functionality. Clean serialization ensures the state converts to JSON or other formats without issues.

Consider what makes sense for your specific use case. An e-commerce assistant might track shopping_cart, order_history, and payment_preferences. A fitness coach agent could maintain workout_plans, progress_metrics, and goals. A customer support bot needs ticket_history, account_info, and interaction_summary. Design your state schema before writing code—it serves as your agent's memory blueprint.

Building Custom Tools with ToolContext

Custom tools transform your agent from a chatbot into a capable assistant with real functionality. Google ADK injects ToolContext into tool functions automatically, providing access to session state. This mechanism enables tools to read and write persistent data seamlessly. Understanding tool development unlocks the full potential of stateful agents.

Let's examine the get_user_state tool implementation:

def get_user_state(tool_context: ToolContext) -> Dict[str, Any]:
    """Retrieve all user information from the session state.

    Args:
        tool_context: Automatically injected by ADK

    Returns:
        dict: User profile including name and preferences
    """
    preferences = tool_context.state.get("preferences", {})

    return {
        "status": "success",
        "preferences": preferences,
        "retrieved_at": datetime.now().isoformat()
    }

This tool retrieves the current preferences from session state and returns them in a structured format. The tool_context.state.get() method safely accesses state with a default fallback. Including metadata like retrieved_at helps with debugging and logging. The return dictionary structure should always be consistent and predictable.

The update_preference tool demonstrates state modification:

def update_preference(
    preference_key: str, 
    preference_value: str, 
    tool_context: ToolContext
) -> Dict[str, Any]:
    """Update or add a user preference to the session state.

    Args:
        preference_key: The preference name (e.g., 'favorite_color', 'favorite_food')
        preference_value: The new preference value
        tool_context: Automatically injected by ADK

    Returns:
        dict: Operation status and updated preferences
    """
    # Get current preferences or initialize empty dict
    preferences = tool_context.state.get("preferences", {})

    # Update the specific preference
    preferences[preference_key] = preference_value

    # Save back to state
    tool_context.state["preferences"] = preferences

    return {
        "status": "success",
        "message": f"Updated {preference_key} to {preference_value}",
        "updated_preferences": preferences,
        "updated_at": datetime.now().isoformat()
    }

This tool accepts parameters that the agent extracts from natural language. When a user says "my favorite food is sushi," the agent calls this tool with preference_key="favorite_food" and preference_value="sushi". The tool retrieves current preferences, modifies the specific key, and saves back to state. Returning the updated preferences confirms the operation succeeded and helps the agent formulate responses.

Tool design best practices include: clear function signatures, comprehensive docstrings, error handling, and informative return values. Clear function signatures tell the agent exactly what parameters to provide. Comprehensive docstrings help both developers and the AI understand tool purpose and usage. Error handling prevents crashes when unexpected inputs occur. Informative return values enable the agent to craft better responses to users.

The ToolContext parameter must always be the last parameter in your function signature. Google ADK automatically injects this parameter—you never pass it explicitly. This convention keeps tool calls clean and focuses on the actual function parameters. The context provides access to state, session metadata, and other runtime information.

Configuring the Agent with Instructions and Tools

Agent configuration determines how your AI assistant behaves and what capabilities it possesses. The instruction prompt shapes the agent's personality, knowledge, and decision-making process.

Tool registration gives the agent access to functions that extend beyond text generation. Getting this configuration right is crucial for building effective persistent session agents.

Here's how we create and configure the agent:

def create_agent(app_name: str) -> Agent:
    """Create and configure the agent with tools and instructions."""
    agent = Agent(
        name=app_name,
        model="gemini-2.0-flash-exp",
        instruction="""
        You are a helpful personal assistant with access to user preferences and state information.

        IMPORTANT GUIDELINES:
        - When a user asks about their preferences, use the get_user_state tool to retrieve their information from the session state.
        - When a user tells you about a new preference or wants to update an existing one, use the update_preference tool to save it to the session state.
        - Provide personalized responses based on the user's stored preferences.

        TOOL USAGE:
        - get_user_state: Retrieves all user preferences
        - update_preference: Updates or adds a new preference (takes preference_key and preference_value)

        Be conversational and helpful. If you don't know something that isn't in the state, say so politely.        
        """,
        tools=[get_user_state, update_preference]
    )
    return agent

The model parameter specifies which Gemini model to use—gemini-2.0-flash-exp provides fast, cost-effective performance. For production applications requiring more sophisticated reasoning, consider gemini-2.0-pro or other variants. Model selection impacts both response quality and operational costs. Choose based on your specific requirements and budget constraints.

The instruction prompt serves as the agent's constitution—it defines how the agent should behave in all situations. Clear instructions about when to use specific tools prevent the agent from making incorrect decisions. Explicit guidelines help the agent understand its capabilities and limitations. The instruction should be conversational yet precise, giving the agent enough context without overwhelming it.

Structure your instructions with sections: identity, capabilities, tool usage, and behavior guidelines. Identity establishes who the agent is and what role it plays. Capabilities outline what the agent can and cannot do. Tool usage provides specific guidance on when and how to use each tool. Behavior guidelines set the tone and approach for interactions.

Tool registration through the tools parameter makes functions available to the agent. The agent receives function signatures and docstrings automatically from Python introspection. This metadata helps the agent understand when to call each tool and what parameters to provide. You can register any number of tools—the agent will choose the appropriate one based on context.

Avoid vague instructions like "be helpful"—instead, specify concrete behaviors. Bad: "Try to remember user preferences." Good: "When a user mentions a preference, immediately call update_preference to store it." Specificity prevents confusion and improves agent reliability.

Implementing the Runner: Orchestrating Agent and Sessions

The Runner component ties everything together, managing the interaction flow between components. It handles message routing, state persistence, and response streaming. Think of the Runner as the conductor of an orchestra, ensuring all parts play in harmony. Understanding Runner mechanics is essential for production-ready agent implementations.

Creating a Runner instance requires three key pieces:

def create_runner(app_name: str, session_service: DatabaseSessionService, agent: Agent) -> Runner:
    """Create a runner to orchestrate the agent and session service."""
    runner = Runner(
        app_name=app_name,
        session_service=session_service,
        agent=agent
    )
    return runner

The app_name identifies your application in the session database—multiple apps can share the same database. The session_service provides the persistence layer for state management. The agent is the configured AI assistant that processes messages and calls tools. These three components work together to create a complete persistent session system.

The Runner exposes two primary methods: run for synchronous execution and run_async for asynchronous streaming. Async execution enables real-time response streaming, improving perceived responsiveness. Users see words appear as the agent generates them, similar to ChatGPT's interface. This streaming approach creates a more engaging user experience than waiting for complete responses.

Here's how we invoke the agent through the Runner:

async def ainvoke_message(runner: Runner, user_id: str, session_id: str, message_text: str):
    """Send a message to the agent and stream the response.

    Args:
        runner: The Runner instance
        user_id: The user ID
        session_id: The session ID
        message_text: The text message to send
    """
    # Create message content
    message = types.Content(
        role="user",
        parts=[types.Part(text=message_text)]
    )

    print("\nAssistant: ", end="", flush=True)

    # Stream agent response
    try:
        async for event in runner.run_async(
            user_id=user_id,
            session_id=session_id,
            new_message=message
        ):
            response = process_agent_event(event)
            if response:
                print(response, end="", flush=True)
    except Exception as e:
        print(f"\nError: {e}")

    print("\n")

The message structure follows Google's Content and Part format for consistency across the AI ecosystem. Each message has a role (user or model) and parts (text, function calls, or function responses). This structure supports multimodal interactions, though our example focuses on text.

The run_async method returns an async generator that yields events as the agent processes the request. Events include thinking steps, tool calls, tool responses, and final text generation. Processing these events lets you build rich user interfaces that show what the agent is doing. The example filters for final responses, but you could display intermediate steps for transparency.

Error handling wraps the streaming loop to catch and report issues gracefully. Production systems should implement more sophisticated error recovery and logging. Consider retry logic for transient failures and detailed error messages for debugging. The Runner automatically persists state changes, even if an error occurs during response generation.

Managing Session Lifecycle: Creation, Retrieval, and Continuation

PS:

If you like this article, share it with others ♻️

Would help a lot ❤️

And feel free to follow me for more content like this.

Session lifecycle management determines how your application handles new and returning users. The lifecycle includes session creation, retrieval, continuation, and eventual cleanup. Proper lifecycle management ensures users have seamless experiences across visits. Let's explore each phase of the session lifecycle in detail.

Session creation initializes state for new users or new conversations:

new_session = await session_service.create_session(
    app_name=APP_NAME,
    user_id=USER_ID,
    state=initial_state()
)
session_id = new_session.id

The create_session method generates a unique session ID and stores the initial state. This ID becomes the primary key for all future interactions within this session. Initial state should provide sensible defaults that make the agent immediately useful. The session service handles all database operations automatically—no SQL required.

Session retrieval checks for existing sessions before creating new ones:

existing_sessions = await session_service.list_sessions(
    app_name=APP_NAME,
    user_id=USER_ID
)

if existing_sessions and len(existing_sessions.sessions) > 0:
    session_id = existing_sessions.sessions[0].id

The list_sessions method returns all sessions for a given app and user combination. In this example, we use the first session found, but production systems might let users choose. Multiple sessions per user enable different conversation contexts—work vs personal, for instance. Session listing helps implement features like conversation history browsing.

Session continuation loads existing state without modifications:

current_session = await session_service.get_session(
    app_name=APP_NAME,
    user_id=USER_ID,
    session_id=session_id
)
if current_session:
    print(f"Current preferences: {current_session.state.get('preferences', {})}")

The get_session method retrieves complete session data including state, metadata, and history. This operation is read-only—it doesn't modify the session or state. Displaying current state at session start helps users understand what the agent remembers. This transparency builds trust in the persistent session system.

Production applications should consider session expiration policies. Inactive sessions older than 90 days might be archived or deleted to manage database size. Implement cleanup jobs that run periodically to remove stale sessions. Balance data retention with privacy concerns and storage costs.

Session locking prevents race conditions in multi-threaded environments. If two requests modify the same session simultaneously, data corruption could occur. Google ADK handles basic concurrency, but high-traffic applications need distributed locks. Consider using Redis or database-level locking for mission-critical session operations.

Building the Interactive Loop: Handling User Input

The interactive loop forms the heart of any conversational AI application. It manages the request-response cycle, handling user input and displaying agent responses. A well-designed loop provides clear feedback, handles edge cases, and enables graceful shutdown. Let's break down the implementation of our interactive loop.

The main loop structure follows a simple but effective pattern:

while True:
    user_input = input("You: ")

    if user_input.lower() in ["exit", "quit"]:
        print("\nSession saved. Have a great day!")

        # Display final state
        final_session = await session_service.get_session(
            app_name=APP_NAME,
            user_id=USER_ID,
            session_id=session_id
        )
        if final_session:
            print(f"Final preferences: {final_session.state.get('preferences', {})}")
        break

    # Skip empty inputs
    if not user_input.strip():
        continue

    # Send message and get response
    await ainvoke_message(runner, USER_ID, session_id, user_input)

The loop uses Python's input() function for simplicity, but production systems need more robust input handling. Web applications would receive input through HTTP requests or WebSocket connections. Mobile apps would collect input through native UI components. The core logic remains the same regardless of input source—receive message, invoke agent, display response.

The message invocation happens asynchronously for responsive streaming:

Async/await syntax enables concurrent operations without blocking the main thread. This approach allows building responsive UIs that remain interactive during agent processing. The await keyword suspends execution until the agent completes its response. Error handling within ainvoke_message prevents the loop from crashing on failures.

Consider adding features like command history (up arrow for previous messages) and auto-completion for better UX. Command history lets users quickly retry or modify previous inputs. Auto-completion helps users discover available commands or preferences. These enhancements transform a basic loop into a polished conversational interface.

Database Configuration and SQLite Integration

Database configuration determines how and where your session state persists. SQLite provides an excellent starting point for development and low-traffic applications. It requires zero configuration, runs in-process, and stores everything in a single file. Understanding database integration helps you scale from prototype to production.

The database URL specifies connection details:

DB_URL = "sqlite:///./session_data.db"

This URL uses the SQLAlchemy format that Google ADK expects. The three slashes after sqlite: indicate a relative file path. The file session_data.db will be created in your current working directory. SQLite creates the database automatically on first use—no setup scripts required.

Initializing the session service connects to the database:

session_service = DatabaseSessionService(db_url=db_url)

The service creates necessary tables on initialization if they don't exist. These tables store session metadata, state, and conversation history. The schema design optimizes for the queries Google ADK performs internally. You don't need to understand the schema details—the abstraction handles everything.

SQLite works great for development, testing, and single-user applications. Its simplicity accelerates initial development and debugging. File-based storage makes backups and version control straightforward. However, SQLite has limitations for production multi-user systems.

Production applications should migrate to PostgreSQL or MySQL for several reasons. These databases handle concurrent writes better than SQLite's locking mechanism. They provide better query optimization for large datasets with thousands of sessions. Cloud-hosted databases integrate with monitoring, backup, and scaling infrastructure. Connection pooling reduces overhead when handling high request volumes.

Migrating from SQLite to PostgreSQL requires only a URL change:

DB_URL = "postgresql://username:password@localhost:5432/session_db"

Google ADK uses SQLAlchemy under the hood, supporting multiple database backends seamlessly. Test your application with your target production database during development. Database-specific behaviors (like JSON handling) can differ between SQLite and PostgreSQL. Load testing with your production database reveals performance characteristics early.

Consider implementing database migrations for schema changes as your application evolves. Tools like Alembic help manage schema versions across deployments. Backup strategies become critical when storing real user data—implement automated daily backups. Monitor database performance metrics to identify slow queries or scaling issues before they impact users.

Advanced State Management Patterns

Beyond basic preference storage, advanced state management patterns enable sophisticated agent behaviors. These patterns help you build agents that handle complex scenarios and maintain rich context. Understanding these techniques elevates your agents from simple bots to intelligent assistants. Let's explore several powerful patterns for state management.

Nested State Structures organize related data hierarchically:

def advanced_initial_state():
    return {
        "profile": {
            "name": "",
            "timezone": "UTC",
            "language": "en"
        },
        "preferences": {
            "theme": "light",
            "notifications": True
        },
        "context": {
            "current_task": None,
            "recent_topics": []
        }
    }

This structure separates concerns into logical groups—profile, preferences, and context. Nested structures prevent flat state from becoming unwieldy as complexity grows. Each subtree can evolve independently without affecting other parts of the state. Tools can target specific subtrees for cleaner, more focused operations.

Time-Stamped History tracks when state changes occurred:

def add_to_history(tool_context: ToolContext, event: str) -> Dict[str, Any]:
    history = tool_context.state.get("history", [])

    history.append({
        "event": event,
        "timestamp": datetime.now().isoformat(),
        "session_id": tool_context.session_id
    })

    # Keep only last 50 events
    tool_context.state["history"] = history[-50:]

    return {"status": "success", "history_length": len(history)}

Time-stamped history enables analytics, debugging, and conversation understanding. The agent can reference what happened "yesterday" or "last week" by examining timestamps. Limiting history size prevents unbounded state growth over long sessions. Consider archiving old history to separate storage if compliance requires complete records.

Computed State Properties derive values from existing state without storing redundancy:

def get_user_summary(tool_context: ToolContext) -> Dict[str, Any]:
    state = tool_context.state
    preferences = state.get("preferences", {})
    profile = state.get("profile", {})

    # Compute derived information
    preference_count = len(preferences)
    is_complete_profile = all(profile.values())

    return {
        "preference_count": preference_count,
        "is_complete_profile": is_complete_profile,
        "user_engaged": preference_count > 5
    }

Computed properties reduce storage costs and prevent inconsistency from duplicate data. The agent gets real-time derived values without worrying about keeping calculations in sync. This pattern works well for metrics, summaries, and business logic evaluations. Complex computations can be memoized if they're expensive and state changes infrequently.

Circuit breakers prevent repeated attempts against failing services. The "open" state blocks requests immediately without waiting for timeouts. The "half-open" state allows testing if the service has recovered. This pattern protects both your application and downstream dependencies from overload.

Mocking ToolContext isolates tool logic from framework dependencies. Each test focuses on a single tool's behavior with known input state. Verify both the return value and state modifications in update operations. These fast unit tests catch bugs early in the development cycle.

Migration tests ensure backward compatibility with existing sessions. They verify that old state formats transform correctly to new schemas. Run these tests whenever you modify state structure. Document expected state transformations for future maintenance.

Code

"""
Persistent Session Agent with Google ADK
A personal assistant that remembers user preferences across conversations.
"""

import asyncio
from datetime import datetime
from typing import Dict, Any, Optional

from google.adk.sessions import DatabaseSessionService
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.tools.tool_context import ToolContext
from google.genai import types

from dotenv import load_dotenv

# Load environment variables from .env file
load_dotenv()

# Constants
APP_NAME = "personal_assistant"
USER_ID = "user_12345"
DB_URL = "sqlite:///./session_data.db"


def get_user_state(tool_context: ToolContext) -> Dict[str, Any]:
    """Retrieve all user information from the session state.

    Args:
        tool_context: Automatically injected by ADK

    Returns:
        dict: User profile including name and preferences
    """
    preferences = tool_context.state.get("preferences", {})

    return {
        "status": "success",
        "preferences": preferences,
        "retrieved_at": datetime.now().isoformat()
    }


def update_preference(
    preference_key: str, 
    preference_value: str, 
    tool_context: ToolContext
) -> Dict[str, Any]:
    """Update or add a user preference to the session state.

    Args:
        preference_key: The preference name (e.g., 'favorite_color', 'favorite_food')
        preference_value: The new preference value
        tool_context: Automatically injected by ADK

    Returns:
        dict: Operation status and updated preferences
    """
    # Get current preferences or initialize empty dict
    preferences = tool_context.state.get("preferences", {})

    # Update the specific preference
    preferences[preference_key] = preference_value

    # Save back to state
    tool_context.state["preferences"] = preferences

    return {
        "status": "success",
        "message": f"Updated {preference_key} to {preference_value}",
        "updated_preferences": preferences,
        "updated_at": datetime.now().isoformat()
    }


def initial_state():
    """Initialize the session state with default preferences."""
    return {
        "preferences": {
            "favorite_color": "blue",
            "favorite_food": "pizza"
        }
    }


def create_agent(app_name: str) -> Agent:
    """Create and configure the agent with tools and instructions."""
    agent = Agent(
        name=app_name,
        model="gemini-2.0-flash-exp",
        instruction="""
        You are a helpful personal assistant with access to user preferences and state information.

        IMPORTANT GUIDELINES:
        - When a user asks about their preferences, use the get_user_state tool to retrieve their information from the session state.
        - When a user tells you about a new preference or wants to update an existing one, use the update_preference tool to save it to the session state.
        - Provide personalized responses based on the user's stored preferences.

        TOOL USAGE:
        - get_user_state: Retrieves all user preferences
        - update_preference: Updates or adds a new preference (takes preference_key and preference_value)

        Be conversational and helpful. If you don't know something that isn't in the state, say so politely.        
        """,
        tools=[get_user_state, update_preference]
    )
    return agent


def create_runner(app_name: str, session_service: DatabaseSessionService, agent: Agent) -> Runner:
    """Create a runner to orchestrate the agent and session service."""
    runner = Runner(
        app_name=app_name,
        session_service=session_service,
        agent=agent
    )
    return runner


def process_agent_event(event) -> Optional[str]:
    """Process agent events and return text to display.

    Args:
        event: Event from the agent runner

    Returns:
        str: Text to display, or None if no display needed
    """
    if event.is_final_response():
        if event.content and event.content.parts:
            return event.content.parts[0].text
    return None


async def ainvoke_message(runner: Runner, user_id: str, session_id: str, message_text: str):
    """Send a message to the agent and stream the response.

    Args:
        runner: The Runner instance
        user_id: The user ID
        session_id: The session ID
        message_text: The text message to send
    """
    # Create message content
    message = types.Content(
        role="user",
        parts=[types.Part(text=message_text)]
    )

    print("\nAssistant: ", end="", flush=True)

    # Stream agent response
    try:
        async for event in runner.run_async(
            user_id=user_id,
            session_id=session_id,
            new_message=message
        ):
            response = process_agent_event(event)
            if response:
                print(response, end="", flush=True)
    except Exception as e:
        print(f"\nError: {e}")

    print("\n")


async def main():
    """Main function to run the persistent session assistant."""
    print("\n" + "=" * 80)
    print("Initializing Persistent Session Agent...")
    print("=" * 80)

    # Initialize session service with database
    db_url = DB_URL
    session_service = DatabaseSessionService(db_url=db_url)

    # Check for existing sessions
    existing_sessions = await session_service.list_sessions(
        app_name=APP_NAME,
        user_id=USER_ID
    )

    # Use existing or create new session
    if existing_sessions and len(existing_sessions.sessions) > 0:
        session_id = existing_sessions.sessions[0].id
        print(f"✓ Continuing session: {session_id}")

        # Get and display current state
        current_session = await session_service.get_session(
            app_name=APP_NAME,
            user_id=USER_ID,
            session_id=session_id
        )
        if current_session:
            print(f"✓ Current preferences: {current_session.state.get('preferences', {})}")
    else:
        new_session = await session_service.create_session(
            app_name=APP_NAME,
            user_id=USER_ID,
            state=initial_state()
        )
        session_id = new_session.id
        print(f"✓ Created new session: {session_id}")
        print(f"✓ Initial preferences: {initial_state()['preferences']}")

    print("\n" + "=" * 80)
    print("Personal Assistant - Type 'exit' or 'quit' to end the conversation")
    print("Try: 'What are my preferences?' or 'I like chocolate'")
    print("=" * 80 + "\n")

    # Create agent and runner
    agent = create_agent(APP_NAME)
    runner = create_runner(APP_NAME, session_service, agent)

    # Main interaction loop
    while True:
        try:
            user_input = input("You: ")

            if user_input.lower() in ["exit", "quit"]:
                print("\n" + "=" * 80)
                print("Session saved. Have a great day!")

                # Display final state
                final_session = await session_service.get_session(
                    app_name=APP_NAME,
                    user_id=USER_ID,
                    session_id=session_id
                )
                if final_session:
                    print(f"✓ Final preferences: {final_session.state.get('preferences', {})}")
                print("=" * 80 + "\n")
                break

            # Skip empty inputs
            if not user_input.strip():
                continue

            # Send message and get response
            await ainvoke_message(runner, USER_ID, session_id, user_input)

        except KeyboardInterrupt:
            print("\n\nInterrupted by user. Saving session...")
            break
        except Exception as e:
            print(f"\nUnexpected error: {e}")
            print("Continuing...\n")


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\nShutting down gracefully...")
    except Exception as e:
        print(f"\nFatal error: {e}")

Conclusion

Persistent sessions transform AI agents from one-shot question-answering systems into intelligent, contextual assistants. Throughout this guide, we've explored the complete architecture of persistent session agents using Google ADK. From state design to tool development, from error handling to production deployment, you now have the knowledge to build sophisticated stateful AI systems.

The key principles we've covered form the foundation of any successful persistent session agent. Design state structure thoughtfully, considering both current needs and future extensibility. Build tools that clearly read and write state while handling errors gracefully.

Configure agents with precise instructions that guide tool usage effectively. Manage session lifecycle carefully, from creation through continuation to cleanup. Test comprehensively across unit, integration, and end-to-end scenarios. Deploy thoughtfully with proper monitoring, security, and scaling considerations.

Google ADK abstracts away much of the complexity traditionally associated with building stateful AI systems. The framework handles serialization, database operations, and state management automatically. This abstraction lets you focus on building great user experiences rather than plumbing infrastructure. The result is agents that remember, learn, and personalize interactions across conversations.

Start simple with basic preference storage, then gradually add sophistication as your requirements evolve. The patterns and techniques covered here scale from prototype to production.

Your agents can grow in complexity while maintaining clean architecture and maintainable code. The future of AI assistants is stateful, contextual, and personalized—and you now have the tools to build it.

PS:

If you like this article, share it with others ♻️

Would help a lot ❤️

And feel free to follow me for more content like this.