Manage sessions with Agent Development Kit

This page describes how you can connect an Agent Development Kit (ADK) agent with Vertex AI Agent Engine Sessions and use managed sessions in the local and production environment.

Before you begin

Make sure your environment is set up by following the Get the required roles and Authentication steps in Set up your environment.

Create a Vertex AI Agent Engine instance

To access Vertex AI Agent Engine Sessions, you first need to create an Vertex AI Agent Engine instance. You don't need to deploy any code to start using Sessions. Without code deployment, creating an Vertex AI Agent Engine instance only takes a few seconds.

import vertexai
from vertexai import agent_engines

# Create an agent engine instance
agent_engine = agent_engines.create()

Develop your ADK agent

To create your ADK agent, follow the instructions in Agent Development Kit, or use the following code to create an agent that greets a user with fixed greetings:

from google import adk

def greetings(query: str):
  """Tool to greet user."""
  if 'hello' in query.lower():
    return {"greeting": "Hello, world"}
  else:
    return {"greeting": "Goodbye, world"}

# Define an ADK agent
root_agent = adk.Agent(
    model="gemini-2.0-flash",
    name='my_agent',
    instruction="You are an Agent that greet users, always use greetings tool to respond.",
    tools=[greetings]
)

Set up the ADK runner

The ADK Runtime orchestrates the execution of your agents, tools, and callbacks, and orchestrates calls to read and write sessions. Initialize the Runner with VertexAiSessionService, which connects with Vertex AI Agent Engine Sessions.

from google.adk.sessions import VertexAiSessionService

app_name="AGENT_ENGINE_ID"
user_id="USER_ID"

# Create the ADK runner with VertexAiSessionService
session_service = VertexAiSessionService(
       "PROJECT_ID", "LOCATION")
runner = adk.Runner(
    agent=root_agent,
    app_name=app_name,
    session_service=session_service)

# Helper method to send query to the runner
def call_agent(query, session_id, user_id):
  content = types.Content(role='user', parts=[types.Part(text=query)])
  events = runner.run(
      user_id=user_id, session_id=session_id, new_message=content)

  for event in events:
      if event.is_final_response():
          final_response = event.content.parts[0].text
          print("Agent Response: ", final_response)

Replace the following:

• PROJECT_ID: Your project ID.

• LOCATION: Your region.

• AGENT_ENGINE_ID: The resource ID of a Vertex AI Agent Engine instance.

  • For deployed agents, the resource ID is listed as the GOOGLE_CLOUD_AGENT_ENGINE_ID environment variable

  • For local agents, you can retrieve the resource ID using agent_engine.name.split("/")[-1].

• USER_ID: A non-empty unique identifier for the user, with a maximum length of 128 characters.

Interact with your agent

After defining your agent and setting up Vertex AI Agent Engine Sessions, you can interact with your agent to check that the session history and states persist.

agent_engine_id="AGENT_ENGINE_ID"

adk web --session_db_url=agentengine://${agent_engine_id}

# Sample output
+-----------------------------------------------------------------------------+
| ADK Web Server started                                                      |
|                                                                             |
| For local testing, access at http://localhost:8000.                         |
+-----------------------------------------------------------------------------+

INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)

# Create a session
session = await session_service.create_session(
       app_name=app_name,
       user_id=user_id)

call_agent("Hello!", session.id, user_id)
# Agent response: "Hello, world"

call_agent("Thanks!", session.id, user_id)
# Agent response: "Goodbye, world"

# List sessions
await session_service.list_sessions(app_name=app_name,user_id=user_id)

# ListSessionsResponse(session_ids=['1122334455', '9988776655'])

# Create a session with state
session = await session_service.create_session(
      app_name=app_name,
      user_id=user_id,
      state={'key': 'value'})

print(session.state['key'])
# value

from google.adk.events import Event, EventActions
import time

# Define state changes
state_changes = {'key': 'new_value'}

# Create event with actions
actions_with_update = EventActions(state_delta=state_changes)
system_event = Event(
  invocation_id="invocation_id",
  author="system", # Or 'agent', 'tool' etc.
  actions=actions_with_update,
  timestamp=time.time()
)

# Append the event
await session_service.append_event(session, system_event)

# Check updated state
updated_session = await session_service.get_session(
      app_name=app_name,
      user_id=user_id,
      session_id=session.id)
# State is updated to new value
print(updated_session.state['key'])
# new_value

await session_service.delete_session(app_name=app_name, user_id=user_id, session_id=session.id)

Deploy your agent to Vertex AI Agent Engine

After you test your agent locally, you can deploy the agent to production by updating the Vertex AI Agent Engine instance with parameters:

agent_engines.update(resource_name=agent_engine.name, agent_engine=AGENT, requirements=REQUIREMENTS)

Replace the following:

• AGENT: The application that implements the query / stream_query method (for example, AdkApp for an ADK agent). For more information, see Deployment considerations.

Clean up

To clean up all resources used in this project, you can delete the Vertex AI Agent Engine instance along with its child resources:

agent_engine.delete(force=True)

What's next

• Manage sessions using API calls.