Chapter 08
Claude with ADK on Vertex AI Agent Engine
NotebookPython 353 cells
In [ ]python · cell 1
python
# Copyright 2025 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.Claude with ADK on Vertex AI Agent Engine
Share to:
| Author(s) |
|---|
| Ivan Nardini |
Overview
This tutorial demonstrates how to build a production-ready, intelligent agent from the ground up. We will use the Google Agent Development Kit (ADK) to orchestrate a powerful third-party model (Anthropic's Claude), connect it to live data with external tools using Model Context Protocol (MCP), and deploy it as a scalable application on Vertex AI Agent Engine.
Finally, we will grant our agent a persistent, long-term memory using Vertex AI Memory Bank and enable tracing capabilities just using Vertex AI Agent Engine.
What you'll learn
By the end of this tutorial, you will be able to:
- Build and test a local agent using the ADK and a third-party model like Claude.
- Augment an agent with tools to interact with external APIs using MCP (e.g., Reddit).
- Deploy your agent as a scalable, managed application using Vertex AI Agent Engine.
- Enable tracing and long-term memory with Vertex AI Memory Bank and the
PreloadMemoryTool
Get started
Install required packages
Our project relies on several key libraries.
The google-cloud-aiplatform SDK provides the tools for Agent Engine and memory, google-adk is the core agent framework, and anthropic allows us to use Claude models on Vertex AI.
The other packages support the Reddit tool we'll add later.
In [ ]python · cell 7
python
%pip install --upgrade --quiet "google-cloud-aiplatform[adk,agent-engines]>=1.110.0" "anthropic[vertex]>=0.55.0" "google-adk[eval]>=1.4.2" "litellm>=1.73.1" "fastmcp>=0.2.0" "redditwarp>=1.0.0" "dnspython>=2.7.0" "praw>=7.8.1" "vertexai>=1.43.0" "click>=8.2.1" "httpx>=0.28.1" "uvicorn>=0.34.3" "nest-asyncio>=1.6.0" "google-auth>=2.40.3" "aiofiles>=24.1.0"Authenticate your notebook environment (Colab only)
If you're running this notebook on Google Colab, run the cell below to authenticate your environment.
In [ ]python · cell 9
python
import sys
if "google.colab" in sys.modules:
from google.colab import auth
auth.authenticate_user()Set Google Cloud project information
To get started using Vertex AI, you must have an existing Google Cloud project and enable the Vertex AI API. Also you can learn more about setting up a project and a development environment.
Let's set our project, location, and a unique Cloud Storage bucket name. This bucket will be used by Vertex AI for staging artifacts during deployment.
In [ ]python · cell 11
python
# Use the environment variable if the user doesn't provide Project ID.
import os
import vertexai
GOOGLE_CLOUD_PROJECT = "[your-project-id]" # @param {type: "string", placeholder: "[your-project-id]", isTemplate: true}
if not GOOGLE_CLOUD_PROJECT or GOOGLE_CLOUD_PROJECT == "[your-project-id]":
GOOGLE_CLOUD_PROJECT = str(os.environ.get("GOOGLE_CLOUD_PROJECT"))
GOOGLE_CLOUD_LOCATION = "europe-west1"
GOOGLE_CLOUD_BUCKET_NAME = "[your-bucket-name]" # @param {type: "string", placeholder: "[your-bucket-name]", isTemplate: true}
GOOGLE_CLOUD_BUCKET = f"gs://{GOOGLE_CLOUD_BUCKET_NAME}"
# Create the bucket if it doesn't exist
! gsutil mb -l $GOOGLE_CLOUD_LOCATION -p $GOOGLE_CLOUD_PROJECT $GOOGLE_CLOUD_BUCKET
# Initialize Vertex AI client
vertexai.init(
project=GOOGLE_CLOUD_PROJECT,
location=GOOGLE_CLOUD_LOCATION,
staging_bucket=GOOGLE_CLOUD_BUCKET,
)Import libraries
In [ ]python · cell 13
python
import os
import logging
logging.getLogger("google_adk").setLevel(logging.CRITICAL)
logging.getLogger("asyncio").setLevel(logging.CRITICAL)
import uuid
from typing import Any, Iterator, Optional
import aiofiles
from google.adk.agents import LlmAgent
from google.adk.models.anthropic_llm import Claude
from google.adk.models.registry import LLMRegistry
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools.mcp_tool import StdioConnectionParams
from google.adk.tools.mcp_tool.mcp_toolset import (MCPToolset,
StdioServerParameters)
from google.genai import types
from vertexai import agent_engines
from vertexai.agent_engines import AgentEngine
from vertexai.preview.reasoning_engines import AdkAppSet variables
In [ ]python · cell 15
python
# Model Selection
MODEL_ID = "claude-sonnet-4@20250514"
# (Optional) Set Reddit Credentials
REDDIT_CLIENT_ID = "[your-reddit-client-id]" # @param {type: "string", placeholder: "[your-reddit-client-id]", isTemplate: true}
REDDIT_CLIENT_SECRET = "[your-reddit-client-secret]" # @param {type: "string", placeholder: "[your-reddit-client-secret]", isTemplate: true}
REDDIT_REFRESH_TOKEN = "[your-reddit-refresh-token]" # @param {type: "string", placeholder: "[your-reddit-refresh-token]", isTemplate: true}
os.environ["MODEL_ID"] = MODEL_ID
os.environ["GOOGLE_CLOUD_PROJECT"] = GOOGLE_CLOUD_PROJECT
os.environ["GOOGLE_CLOUD_LOCATION"] = GOOGLE_CLOUD_LOCATION
if not REDDIT_CLIENT_ID or REDDIT_CLIENT_ID == "[your-reddit-client-id]" or not REDDIT_CLIENT_SECRET or REDDIT_CLIENT_SECRET == "[your-reddit-client-secret]" or not REDDIT_REFRESH_TOKEN or REDDIT_REFRESH_TOKEN == "[your-reddit-refresh-token]":
REDDIT_CLIENT_ID = str(os.environ.get("REDDIT_CLIENT_ID"))
REDDIT_CLIENT_SECRET = str(os.environ.get("REDDIT_CLIENT_SECRET"))
REDDIT_REFRESH_TOKEN = str(os.environ.get("REDDIT_REFRESH_TOKEN"))Helpers
This tutorial uses a helper function, chat_loop, to provide a consistent interactive chat interface for testing the agent in both local and deployed environments. This function abstracts the differences between interacting with a local ADK Runner and a remote AgentEngine instance.
In [ ]python · cell 17
python
def chat_loop(
app, user_id: Optional[str] = None, session_id: Optional[str] = None
) -> None:
"""Interactive chat loop for AI applications."""
# Simple setup
user_id = user_id or f"u_{uuid.uuid4().hex[:8]}"
# Handle session based on app type
if isinstance(app, (AdkApp, AgentEngine)):
# Only create session if session_id is not provided
if not session_id:
session = app.create_session(user_id=user_id)
# Handle both dict and object session responses
if isinstance(session, dict):
session_id = session["id"]
else:
session_id = session.id
def query_fn(msg: str):
return app.stream_query(user_id=user_id, session_id=session_id, message=msg)
elif isinstance(app, Runner):
session_id = session_id or f"s_{uuid.uuid4().hex[:8]}"
def query_fn(msg: str):
return app.run(
user_id=user_id,
session_id=session_id,
new_message=types.Content(role="user", parts=[types.Part(text=msg)]),
)
else:
raise TypeError(
f"Unsupported app type: {type(app)}. Expected AdkApp, AgentEngine, or Runner."
)
_print_startup_info(user_id, session_id)
# Main loop
while True:
try:
user_input = input("\nYou: ").strip()
if not user_input or user_input.lower() in {"quit", "exit", "bye"}:
break
print("\nAssistant: ", end="", flush=True)
response = _get_response_text(query_fn(user_input))
print(response or "(No response generated)")
except (KeyboardInterrupt, EOFError):
print("\n\n🛑 Chat interrupted.")
break
except Exception as e:
print(f"\n❌ Error: {e}")
continue
print("\n👋 Goodbye!")
def _print_startup_info(user_id: str, session_id: str) -> None:
"""Print startup information."""
print("\n🚀 Starting chat...")
print(f"👤 User ID: {user_id}")
print(f"📁 Session ID: {session_id}")
print("💬 Type 'exit' or 'quit' to end.")
print("-" * 50)
def _get_response_text(events: Iterator[Any]) -> str:
"""Extract response text from event stream."""
responses = []
for event in events:
# Handle dict-like events (AgentEngine format)
if isinstance(event, dict):
text = _extract_from_dict_event(event)
# Handle object-like events
else:
text = _extract_from_object_event(event)
if text:
responses.append(text)
# Print streaming text in real-time
print(text, end="", flush=True)
return "".join(responses)
def _extract_from_dict_event(event: dict) -> Optional[str]:
"""Extract text from dictionary-style events (AgentEngine format)."""
# Handle AgentEngine response format
if "parts" in event and "role" in event:
# Only extract text from model responses, skip function calls/responses
if event.get("role") == "model":
parts = event.get("parts", [])
text_parts = []
for part in parts:
# Extract text content, skip function calls
if isinstance(part, dict) and "text" in part:
text_parts.append(part["text"])
return "".join(text_parts) if text_parts else None
return None
# Handle other dict formats
content = event.get("content", {})
if isinstance(content, str):
return content
parts = content.get("parts", [])
if not parts:
return None
text_parts = []
for part in parts:
if isinstance(part, dict) and "text" in part:
text_parts.append(part["text"])
return "".join(text_parts) if text_parts else None
def _extract_from_object_event(event: Any) -> Optional[str]:
"""Extract text from object-style events."""
# Handle string content directly
content = getattr(event, "content", None)
if isinstance(content, str):
return content
# Handle content with parts
if content and hasattr(content, "parts"):
text_parts = []
for part in content.parts:
if hasattr(part, "text") and part.text:
text_parts.append(part.text)
return "".join(text_parts) if text_parts else None
# Handle direct text attribute
return getattr(event, "text", None)Step 1: Create a simple ADK Social Media Agent
Let's start with the simplest possible agent: a creative assistant that can brainstorm social media content.
Define your 1st ADK agent
The core of our application is the LlmAgent. Think of it as the agent's brain. We define its personality and purpose through its instruction (also known as a system prompt) and specify which LLM it should use.
In [ ]python · cell 20
python
# Register Claude Model for ADK
LLMRegistry.register(Claude)
# Define the Social Media Assistant Agent
root_agent = LlmAgent(
name="SocialMediaAssistant",
model=os.getenv("MODEL_ID", "claude-sonnet-4@20250514"),
description="Helps create engaging social media content and strategies.",
instruction=(
"You are a creative and knowledgeable Social Media Assistant.\n"
"Your goal is to help users create engaging social media content and develop strategies.\n"
"1. Help users brainstorm content ideas for various social media platforms (Instagram, Twitter, LinkedIn, etc.).\n"
"2. Provide suggestions for captions, hashtags, and post timing.\n"
"3. Offer advice on content strategy, audience engagement, and brand voice.\n"
"4. Help with content planning and scheduling recommendations.\n"
"5. Suggest trending topics and content formats that might work well.\n"
"6. Provide best practices for different social media platforms.\n"
"7. Maintain a professional yet creative tone throughout the conversation.\n"
"8. Ask clarifying questions about the user's target audience, platform preferences, and content goals when needed."
),
tools=[],
)Test the simple agent locally
With our agent defined, let's chat with it. For local development, the ADK provides in-memory services (InMemorySessionService) that simulate a production environment without any cloud dependencies. This is perfect for rapid iteration and testing.
The Runner is the engine that orchestrates the conversation, managing the flow between the user, the agent, and any services. We'll use a simple chat_loop helper to make the interaction feel natural.
In [ ]python · cell 22
python
async def test_agent_locally():
# Create the session service
session_service = InMemorySessionService()
# Create the specific session where the conversation will happen
user_id = "user_01"
session = await session_service.create_session(
app_name="MyApp", user_id=user_id
)
# Initialize the Runner with the correct session service
runner = Runner(
agent=root_agent,
app_name="MyApp",
session_service=session_service,
)
try:
chat_loop(runner, user_id, session.id)
except KeyboardInterrupt:
print("\n\n🛑 Chat interrupted.")
except Exception as e:
print(f"\n❌ Error: {e}")You now have a running, simple, agent powered by Claude. Try asking it for some social media post ideas!
In [ ]python · cell 24
python
await test_agent_locally()Step 2: Augmenting the agent with tools using MCP
Our agent is creative, but it's isolated. To be truly useful, it needs to interact with the outside world. We'll give it the ability to research topics on Reddit.
Tools are how agents perform actions. The ADK uses the Model Context Protocol (MCP), a standard for building and connecting tools. We'll use a pre-built MCP tool for Reddit.
Prepare the tool's runtime environment
The Reddit tool is an external Python package. To install it, we can use a simple shell script.
In [ ]python · cell 27
python
!mkdir -p installation_scriptsIn [ ]python · cell 28
python
install_local_mcp_file = """
#!/bin/bash
# Exit immediately if a command exits with a non-zero status.
set -e
echo "Installing MCP Reddit Server"
# Install uv (a fast Python package manager)
apt-get update
apt-get install -y curl
echo "Installing uv..."
curl -LsSf https://astral.sh/uv/install.sh | sh
# Add uv to PATH for current session
export PATH="$HOME/.local/bin:$PATH"
# Install the mcp-reddit tool using the command from its documentation
echo "Installing mcp-reddit using uv..."
uv pip install "git+https://github.com/adhikasp/mcp-reddit.git" --system
echo "MCP Reddit Server installation complete."
"""
with open("installation_scripts/install_local_mcp.sh", "w") as f:
f.write(install_local_mcp_file)
f.close()In [ ]python · cell 29
python
!chmod +x installation_scripts/install_local_mcp.sh && ./installation_scripts/install_local_mcp.shUpdate the agent definition
Now, we update our LlmAgent to include the MCPToolset. We also refine its instructions, explicitly telling it about its new capabilities. This is critical—the LLM needs to know what tools it has and when to use them.
Notice: To solve the
filenoerror in Colab, we must redirect the tool's error stream to a file that supports asynchronous operations. We will use the aiofiles library for this. This requires our test to run inside an async function.
In [ ]python · cell 31
python
def create_agent(errlog):
root_agent = LlmAgent(
name="SocialMediaAssistantWithReddit",
model=os.getenv("MODEL_ID", "claude-sonnet-4@20250514"),
description="Helps create engaging social media content with Reddit research capabilities.",
instruction=(
"You are a creative Social Media Assistant with Reddit research capabilities.\n"
"Your goal is to help users create engaging social media posts for LinkedIn and X.\n\n"
"You have access to Reddit tools that can help you:\n"
"1. **fetch_hot_threads**: Get hot/trending threads from any subreddit\n"
"2. **fetch_post_content**: Get detailed post content including comments\n\n"
"Use these tools to gather the latest trends and create data-driven social media content."
),
tools=[
MCPToolset(
connection_params=StdioConnectionParams(
server_params=StdioServerParameters(
command="mcp-reddit",
),
),
errlog=errlog, # Required only in colab environment
)
],
)
return root_agentTest the agent locally
Now, try a prompt like: "What are some hot topics in the r/MachineLearning subreddit?" You should see the agent activate its tool to fetch live data from Reddit!
In [ ]python · cell 33
python
async def test_agent_locally():
# Create the session service
session_service = InMemorySessionService()
# Create the specific session where the conversation will happen
user_id = "user_01"
session = await session_service.create_session(
app_name="MyApp", user_id=user_id
)
# Initialize the Runner with the correct session service
errlog = await aiofiles.open(
"error.log", "w+"
) # Required only in colab environment.
root_agent = create_agent(errlog)
runner = Runner(
agent=root_agent,
app_name="MyApp",
session_service=session_service
)
try:
chat_loop(runner, user_id, session.id)
finally:
# Ensure the log file is always closed
await errlog.close()In [ ]python · cell 34
python
await test_agent_locally()Step 3: Scaling up to Vertex AI Agent Engine
Local testing is fantastic for development speed, but real applications need to be deployed. Vertex AI Agent Engine is a managed service that hosts, scales, and monitors your agent, turning your Python script into a production-ready API.
Create the agent module
To deploy, we must package our agent's definition into a Python file, which we'll call root_agent.py. This file will contain not only the agent logic but also "builders" for cloud services like VertexAiSessionService, the production equivalent of InMemorySessionService.
Notice we wrap our agent definition in an AdkApp. This is the entry point that Agent Engine will use to serve our agent.
In [ ]python · cell 37
python
root_agent_file = f"""
import os
from google.adk.models.anthropic_llm import Claude
from google.adk.models.registry import LLMRegistry
from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, StdioServerParameters
from google.adk.tools.mcp_tool import StdioConnectionParams
import vertexai
from vertexai.preview.reasoning_engines import AdkApp
# Set variable
GOOGLE_CLOUD_PROJECT=os.getenv('GOOGLE_CLOUD_PROJECT', '{GOOGLE_CLOUD_PROJECT}')
GOOGLE_CLOUD_LOCATION=os.getenv('GOOGLE_CLOUD_LOCATION', '{GOOGLE_CLOUD_LOCATION}')
MODEL_ID=os.getenv('MODEL_ID', '{MODEL_ID}')
REDDIT_CLIENT_ID=os.getenv('REDDIT_CLIENT_ID', '{REDDIT_CLIENT_ID}')
REDDIT_CLIENT_SECRET=os.getenv('REDDIT_CLIENT_SECRET', '{REDDIT_CLIENT_SECRET}')
REDDIT_REFRESH_TOKEN=os.getenv('REDDIT_REFRESH_TOKEN', '{REDDIT_REFRESH_TOKEN}')
# Initialize Vertex AI SDK
vertexai.init(project=GOOGLE_CLOUD_PROJECT, location=GOOGLE_CLOUD_LOCATION)
# Register Claude Model for ADK
LLMRegistry.register(Claude)
# Define session builder
def session_service_builder():
# This is needed to ensure InitGoogle and AdkApp setup is called first.
from google.adk.sessions import VertexAiSessionService
return VertexAiSessionService(project=GOOGLE_CLOUD_PROJECT, location=GOOGLE_CLOUD_LOCATION)
agent_app = AdkApp(
agent=LlmAgent(
name='SocialMediaAssistantWithReddit',
model=os.getenv('MODEL_ID'),
description='Helps create engaging social media content with Reddit research capabilities.',
instruction=(
'You are a creative Social Media Assistant with Reddit research capabilities.\\n'
'Your goal is to help users create engaging social media posts for LinkedIn and X.\\n\\n'
'You have access to Reddit tools that can help you:\\n'
'1. **fetch_hot_threads**: Get hot/trending threads from any subreddit\\n'
'2. **fetch_post_content**: Get detailed post content including comments\\n\\n'
'Use these tools to gather the latest trends and create data-driven social media content.'
),
tools=[
MCPToolset(
connection_params=StdioConnectionParams(
server_params=StdioServerParameters(
command='mcp-reddit',
),
timeout=60,
),
)
],
),
session_service_builder=session_service_builder,
)
"""
with open("root_agent.py", "w") as f:
f.write(root_agent_file)
f.close()Deploy the Agent
Now we call agent_engines.create(). This command packages our code, installs dependencies, and deploys it to a secure, scalable environment.
agent_engine: We use aModuleAgentto point to ourroot_agent.pyfile and theagent_appobject within it.extra_packages: We include our agent module and the installation script.env_vars: We securely pass our Reddit API keys to the production environment.build_options: We instruct the build process to run our custom installation script.
In [ ]python · cell 39
python
remote_app = agent_engines.create(
display_name="reddit_assistant_agent",
description="A Reddit assistant agent with MCP",
agent_engine=agent_engines.ModuleAgent(
module_name="root_agent",
agent_name="agent_app",
# You can get the register operations with app = AdkApp(root_agent, session_service_builder), app.setup, app.register_operations()
register_operations={'': ['get_session', 'list_sessions', 'create_session', 'delete_session'],
'async': ['async_get_session',
'async_list_sessions',
'async_create_session',
'async_delete_session',
'async_add_session_to_memory',
'async_search_memory'],
'stream': ['stream_query', 'streaming_agent_run_with_events'],
'async_stream': ['async_stream_query']}
),
# Python dependencies for the cloud environment.
requirements=[
"google-cloud-aiplatform[adk,agent-engines]>=1.110.0",
"anthropic[vertex]>=0.55.0",
"litellm>=1.73.1",
"fastmcp>=0.2.0",
"redditwarp>=1.0.0",
"dnspython>=2.7.0",
"praw>=7.8.1",
"python-dotenv>=1.0.0",
"vertexai>=1.43.0",
],
# Files to include in the deployment package.
extra_packages=[
"root_agent.py",
"installation_scripts/install_local_mcp.sh",
],
# Environment variables for the deployed agent.
env_vars={
"MODEL_ID" : MODEL_ID,
"REDDIT_CLIENT_ID": REDDIT_CLIENT_ID,
"REDDIT_CLIENT_SECRET": REDDIT_CLIENT_SECRET,
"REDDIT_REFRESH_TOKEN": REDDIT_REFRESH_TOKEN,
},
# Custom build steps.
build_options={
"installation": [
"installation_scripts/install_local_mcp.sh",
],
},
)Test the agent remotely
Once deployed, you can interact with the remote_app object just like you did with the local Runner. The interface is identical, but the execution is now happening on Google Cloud.
In [ ]python · cell 41
python
user_id = "user_01"
session = remote_app.create_session(user_id=user_id)
chat_loop(remote_app, user_id=user_id, session_id=session['id'])Generate memories from the previous conversation
You have two methods to add memory from previous conversation or session. You can use:
generate_memoriesmethodasync_add_session_to_memorymethod
In this case, we use the generate_memories method to better control.
In [ ]python · cell 43
python
agent_engine_name = remote_app.resource_name
app_name = remote_app.display_name
response = agent_engines.generate_memories(
name=agent_engine_name,
vertex_session_source={
"session": f"{agent_engine_name}/sessions/{session['id']}"
},
scope= {
"user_id": user_id,
"app_name": app_name
}
)In [ ]python · cell 44
python
if response.response.generated_memories:
for id, m in enumerate(response.response.generated_memories):
memory = agent_engines.get_memory(name=m.memory.name)
print("=========================")
print(f"Memory {id}")
print("=========================")
print(memory.scope)
print(memory.fact)
print(memory)
print("\n")
else:
print("No generated memories found.")(Optional) Step 4: Enable memory and tracing capabilities with your agent
Update the agent module
We'll update our root_agent.py to include two new components:
VertexAiMemoryBankService: The service that connects our agent to the Memory Bank backend.PreloadMemoryTool: A built-in ADK tool that automatically queries the Memory Bank at the start of each turn, retrieving relevant facts about the user.
We simply add a memory_service_builder and include the PreloadMemoryTool in our agent's tool list. We'll also set enable_tracing=True for better observability in the Google Cloud console.
In [ ]python · cell 47
python
agent_engine_id = remote_app.resource_name.split("/")[-1]
root_agent_file = f"""
import os
from google import adk
from google.adk.models.anthropic_llm import Claude
from google.adk.models.registry import LLMRegistry
from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, StdioServerParameters
from google.adk.tools.mcp_tool import StdioConnectionParams
import vertexai
from vertexai.preview.reasoning_engines import AdkApp
# Set variable
GOOGLE_CLOUD_PROJECT=os.getenv('GOOGLE_CLOUD_PROJECT', '{GOOGLE_CLOUD_PROJECT}')
GOOGLE_CLOUD_LOCATION=os.getenv('GOOGLE_CLOUD_LOCATION', '{GOOGLE_CLOUD_LOCATION}')
GOOGLE_AGENT_ENGINE_ID=os.getenv('GOOGLE_AGENT_ENGINE_ID', '{agent_engine_id}')
MODEL_ID=os.getenv('MODEL_ID', '{MODEL_ID}')
REDDIT_CLIENT_ID=os.getenv('REDDIT_CLIENT_ID', '{REDDIT_CLIENT_ID}')
REDDIT_CLIENT_SECRET=os.getenv('REDDIT_CLIENT_SECRET', '{REDDIT_CLIENT_SECRET}')
REDDIT_REFRESH_TOKEN=os.getenv('REDDIT_REFRESH_TOKEN', '{REDDIT_REFRESH_TOKEN}')
# Initialize Vertex AI SDK
vertexai.init(project=GOOGLE_CLOUD_PROJECT, location=GOOGLE_CLOUD_LOCATION)
# Register Claude Model for ADK
LLMRegistry.register(Claude)
# Define session builder
def session_service_builder():
from google.adk.sessions import VertexAiSessionService
return VertexAiSessionService(project=GOOGLE_CLOUD_PROJECT, location=GOOGLE_CLOUD_LOCATION)
# Define memory builder
def memory_service_builder():
from google.adk.memory import VertexAiMemoryBankService
return VertexAiMemoryBankService(project=GOOGLE_CLOUD_PROJECT, location=GOOGLE_CLOUD_LOCATION, agent_engine_id=GOOGLE_AGENT_ENGINE_ID)
agent_app = AdkApp(
agent=LlmAgent(
name='SocialMediaAssistantWithReddit',
model=os.getenv('MODEL_ID'),
description='Helps create engaging social media content with Reddit research capabilities.',
instruction=(
'You are a creative Social Media Assistant with Reddit research capabilities.\\n'
'Your goal is to help users create engaging social media posts for LinkedIn and X.\\n\\n'
'You have access to Reddit tools that can help you:\\n'
'1. **fetch_hot_threads**: Get hot/trending threads from any subreddit\\n'
'2. **fetch_post_content**: Get detailed post content including comments\\n\\n'
'Use these tools to gather the latest trends and create data-driven social media content.'
),
tools=[
MCPToolset(
connection_params=StdioConnectionParams(
server_params=StdioServerParameters(
command='mcp-reddit',
),
timeout=60,
),
),
adk.tools.preload_memory_tool.PreloadMemoryTool()
]
),
session_service_builder=session_service_builder,
memory_service_builder=memory_service_builder,
enable_tracing=True
)
"""
with open("root_agent.py", "w") as f:
f.write(root_agent_file)
f.close()Update the deployed agent
Instead of creating a new agent, we use agent_engines.update() to deploy our new version. The process is the same, but it updates the existing endpoint.
In [ ]python · cell 49
python
remote_app = agent_engines.update(
resource_name=agent_engine_name,
agent_engine=agent_engines.ModuleAgent(
module_name="root_agent",
agent_name="agent_app",
register_operations={'': ['get_session', 'list_sessions', 'create_session', 'delete_session'],
'async': ['async_get_session',
'async_list_sessions',
'async_create_session',
'async_delete_session',
'async_add_session_to_memory',
'async_search_memory'],
'stream': ['stream_query', 'streaming_agent_run_with_events'],
'async_stream': ['async_stream_query']}
),
requirements=[
"google-cloud-aiplatform[adk,agent-engines]>=1.110.0",
"anthropic[vertex]>=0.55.0",
"litellm>=1.73.1",
"fastmcp>=0.2.0",
"redditwarp>=1.0.0",
"dnspython>=2.7.0",
"praw>=7.8.1",
"python-dotenv>=1.0.0",
"vertexai>=1.43.0",
"aiofiles>=24.1.0"
],
extra_packages=[
"root_agent.py",
"installation_scripts/install_local_mcp.sh",
],
env_vars={
"MODEL_ID" : MODEL_ID,
"REDDIT_CLIENT_ID": REDDIT_CLIENT_ID,
"REDDIT_CLIENT_SECRET": REDDIT_CLIENT_SECRET,
"REDDIT_REFRESH_TOKEN": REDDIT_REFRESH_TOKEN,
"GOOGLE_AGENT_ENGINE_ID" : agent_engine_id,
},
build_options={
"installation": [
"installation_scripts/install_local_mcp.sh",
],
},
)Test the agent remotely
Once updated, you can interact as before. But in this case the agent should remember details about previous conversation.
In [ ]python · cell 51
python
user_id = "user_01"
session = remote_app.create_session(user_id=user_id)
chat_loop(remote_app, user_id=user_id, session_id=session['id'])Cleaning up
To avoid incurring future charges, it's important to delete the resources we've created. This command will delete the Agent Engine deployment and the associated services. We also remove the Cloud Storage bucket.
In [ ]python · cell 53
python
delete_agent_engine = True
delete_bucket = True
if delete_agent_engine:
agent_engine_name = remote_app.delete(force=True)
if delete_bucket:
!gsutil rm -r $GOOGLE_CLOUD_BUCKET