Generative AI on Google Cloud

15 / 80·notebook

Chapter 15

MCP on Vertex AI Agent Engine with custom installation scripts

NotebookPython 337 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.

MCP on Vertex AI Agent Engine with custom installation scripts

Share to:

Author(s)
Shawn Yang, Ivan Nardini

Overview

Vertex AI Agent Engine provides support for custom installation scripts. This means you can now run shell scripts you need during your agent's build process. You can install custom binaries, compile code from source, or set up any other specialized dependency your agent requires.

Most importantly, this feature means unlocks a new pattern for deploying agents with complex, standardized tools. In fact, you can now deploy agents using the Model Context Protocol (MCP) on Vertex AI Agent Engine.

This guide shows you how to build, test, and deploy a custom Reddit agent that uses an MCP server, demonstrating this new pattern for creating custom agents on Vertex AI Agent Engine.

What You'll Learn

How to set up your local environment for agent development.
How to define and configure a toolset using the Model Context Protocol (MCP).
How to build and test an agent locally using the Google Agent Development Kit (ADK).
How to use a single, unified function for interactive testing in both local and remote environments.
How to use the "bring your own installation script" feature to deploy the agent to Vertex AI Agent Engine.

Get started

Install Google Gen AI SDK and other required packages

Run the following command to install the Vertex AI SDK with the necessary extras for Agent Engine and the ADK.

In [ ]python · cell 7

python

%pip install "google-cloud-aiplatform[agent_engines,adk]>=1.101.0" "aiofiles" --force-reinstall --quiet

Important: After the installation completes, you must restart your Colab or notebook kernel for the new packages to be recognized.

Authenticate your notebook environment (Colab only)

Next, authenticate your account and initialize the Vertex AI SDK. This allows your environment to interact with your Google Cloud project.

In [ ]python · cell 10

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.

Learn more about setting up a project and a development environment.

In [ ]python · cell 12

python

# Use the environment variable if the user doesn't provide Project ID.
import os

import vertexai

PROJECT_ID = "[your-project-id]"  # @param {type: "string", placeholder: "[your-project-id]", isTemplate: true}
if not PROJECT_ID or PROJECT_ID == "[your-project-id]":
    PROJECT_ID = str(os.environ.get("GOOGLE_CLOUD_PROJECT"))

LOCATION = os.environ.get("GOOGLE_CLOUD_REGION", "us-central1")

BUCKET_NAME = "[your-bucket-name]"  # @param {type: "string", placeholder: "[your-bucket-name]", isTemplate: true}

BUCKET_URI = f"gs://{BUCKET_NAME}"

! gsutil mb -l {LOCATION} -p {PROJECT_ID} {BUCKET_URI}

# Set environment variables required for ADK
os.environ["GOOGLE_GENAI_USE_VERTEXAI"] = "TRUE"
os.environ["GOOGLE_CLOUD_PROJECT"] = PROJECT_ID
os.environ["GOOGLE_CLOUD_LOCATION"] = LOCATION

# Initialize Vertex AI
vertexai.init(project=PROJECT_ID, location=LOCATION, staging_bucket=BUCKET_URI)

(Optional) Get Reddit API Credentials

Our agent needs credentials to access the Reddit API. You'll need to create a "script" app in your Reddit account preferences to get a client ID and client secret. You will also need a refresh token.

Important Security Note: We will handle these credentials securely using environment variables. Never hardcode secrets directly in your code.

In [ ]python · cell 14

python

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}

Import required libraries

Import the necessary libraries from the ADK, Vertex AI SDK, and Python's standard library.

In [ ]python · cell 16

python

import logging

logging.getLogger("google_adk").setLevel(logging.CRITICAL)
logging.getLogger("asyncio").setLevel(logging.CRITICAL)
import os
import uuid
from typing import Any, Iterator, Optional

import aiofiles
from google.adk.agents import LlmAgent
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 AdkApp

Helper function

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 18

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)

Building the Agent Locally

Before deployment, the agent is built and tested in the local environment. This allows for rapid iteration and debugging.

Prepare the installation script

The agent uses an external tool, mcp-reddit, which must be installed in the runtime environment. A shell script will be created to handle this installation. Vertex AI Agent Engine will execute this script during the deployment process.

First, create a local directory to store the script.

Then, the script performs the following actions:

Updates the package index and installs curl.
Installs uv, a fast Python package manager, to accelerate dependency installation.
Adds uv to the system's PATH.
Uses uv to install the mcp-reddit package directly from its Git repository.

In [ ]python · cell 21

python

!mkdir -p installation_scripts

In [ ]python · cell 22

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 23

python

!chmod +x installation_scripts/install_local_mcp.sh && ./installation_scripts/install_local_mcp.sh

Defining the Agent

Here, we define the LlmAgent using ADK. We provide a detailed prompt and connect it to our MCPToolset, pointing it to the runner script. We define our agent's using a factory function. The main test loop will then call this function to create an agent instance, providing it with the necessary async log file handle at runtime.

Notice: To solve the fileno error 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 25

python

def create_agent(errlog):
    root_agent = LlmAgent(
        model="gemini-2.5-flash",
        name="reddit_assistant_agent",
        instruction="Help the user fetch reddit info.",
        tools=[
            MCPToolset(
                connection_params=StdioConnectionParams(
                    server_params=StdioServerParameters(
                        command="mcp-reddit",
                    ),
                ),
                errlog=errlog,  # Required only in colab environment
            )
        ],
    )
    return root_agent

Test the agent locally

Test the agent on the local machine to verify its functionality before deploying to the cloud.

The local test setup involves:

Instantiating in-memory versions of the SessionService, MemoryService, and ArtifactService for lightweight local testing.
Creating an async file handle for the tool's error log.
Initializing the ADK Runner, which orchestrates the interaction between the user and the agent.
Calling the chat_loop helper to start an interactive conversation with the agent.

In [ ]python · cell 27

python

async def test_agent_locally():

    # Create the session service
    session_service = InMemorySessionService()

    # Create the specific session where the conversation will happen
    user_id = "runner_user_01"
    session = await session_service.create_session(
        app_name="MyRunnerApp", 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="MyRunnerApp", 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 28

python

await test_agent_locally()

Deploying to Vertex AI Agent Engine

With our agent tested and working locally, it's time to deploy our MCP-enabled agent.

Create the agent module

Since ADK 1.0.0, the MCPToolset contains non-serializable (non-pickleable) state, such as thread locks.

To deploy an agent with such a toolset, it must be wrapped in a ModuleAgent. This approach involves defining the agent in a separate Python file (root_agent.py) and referencing it by the module and variable name during deployment.

This agent module file defines the full application, including service builders for VertexAiSessionService, which are the cloud-based counterparts to the in-memory services used for local testing. The agent itself is wrapped in an AdkApp object.

In [ ]python · cell 31

python

root_agent_file = f"""
import os
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
from vertexai.preview.reasoning_engines import AdkApp

# Set variable
PROJECT_ID=os.getenv('PROJECT_ID', '{PROJECT_ID}')
LOCATION=os.getenv('LOCATION', '{LOCATION}')
BUCKET_NAME=os.getenv('BUCKET_NAME', '{BUCKET_NAME}')
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}')

# 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=PROJECT_ID, location=LOCATION)

agent_app = AdkApp(
    agent=LlmAgent(
        model='gemini-2.5-flash',
        name='reddit_assistant_agent',
        instruction='Help the user fetch reddit info.',
        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

The agent_engines.create() function is used to deploy the agent. The configuration specifies several key parameters:

agent_engine: An agent_engines.ModuleAgent is used, pointing to the root_agent.py file (module_name) and the agent_app variable within it (agent_name).
requirements: A list of standard Python package dependencies.
extra_packages: A list of local files to be included in the build. This must include the agent module file (root_agent.py) and the custom installation script.
env_vars: A dictionary of environment variables to be set in the deployed container, used here to securely pass the Reddit API credentials.
build_options: This dictionary specifies custom build-time operations. The installation key is used to provide a list of scripts that will be executed before the application server starts. This is where the custom installation script is specified.

In [ ]python · cell 33

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",
        register_operations={
            "": ["get_session", "list_sessions", "create_session", "delete_session"],
            "async": [
                "async_get_session",
                "async_list_sessions",
                "async_create_session",
                "async_delete_session",
            ],
            "stream": ["stream_query", "streaming_agent_run_with_events"],
            "async_stream": ["async_stream_query"],
        },
    ),
    requirements=["google-cloud-aiplatform[agent_engines,adk]>=1.101.0"],
    extra_packages=[
        "root_agent.py",
        "installation_scripts/install_local_mcp.sh",
    ],
    env_vars={
        "PROJECT_ID": PROJECT_ID,
        "LOCATION": LOCATION,
        "REDDIT_CLIENT_ID": REDDIT_CLIENT_ID,
        "REDDIT_CLIENT_SECRET": REDDIT_CLIENT_SECRET,
        "REDDIT_REFRESH_TOKEN": REDDIT_REFRESH_TOKEN,
    },
    build_options={
        "installation": [
            "installation_scripts/install_local_mcp.sh",
        ],
    },
)

After deployment, the chat_loop function can be used again to interact with the now remotely-hosted agent.

In [ ]python · cell 35

python

chat_loop(remote_app)

Cleaning up

To avoid incurring ongoing charges to your Google Cloud account for the resources used in this tutorial, delete the resources you created. This section provides commands to delete the deployed Agent Engine and the associated Cloud Storage bucket.

In [ ]python · cell 37

python

delete_agent_engine = True
delete_bucket = True

if delete_agent_engine:
    agent_engines = agent_engines.list(filter="display_name=reddit_assistant_agent")
    for agent_engine in agent_engines:
        agent_engine.delete(force=True)

if delete_bucket:
    !gsutil rm -r {BUCKET_URI}

# 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.

Author(s)

Shawn Yang, Ivan Nardini

Overview

This guide shows you how to build, test, and deploy a custom Reddit agent that uses an MCP server, demonstrating this new pattern for creating custom agents on Vertex AI Agent Engine.

What You'll Learn

How to set up your local environment for agent development.

How to define and configure a toolset using the Model Context Protocol (MCP).

How to build and test an agent locally using the Google Agent Development Kit (ADK).

How to use a single, unified function for interactive testing in both local and remote environments.

How to use the "bring your own installation script" feature to deploy the agent to Vertex AI Agent Engine.

# Use the environment variable if the user doesn't provide Project ID. import os import vertexai PROJECT_ID = "[your-project-id]" # @param {type: "string", placeholder: "[your-project-id]", isTemplate: true} if not PROJECT_ID or PROJECT_ID == "[your-project-id]": PROJECT_ID = str(os.environ.get("GOOGLE_CLOUD_PROJECT")) LOCATION = os.environ.get("GOOGLE_CLOUD_REGION", "us-central1") BUCKET_NAME = "[your-bucket-name]" # @param {type: "string", placeholder: "[your-bucket-name]", isTemplate: true} BUCKET_URI = f"gs://{BUCKET_NAME}" ! gsutil mb -l {LOCATION} -p {PROJECT_ID} {BUCKET_URI} # Set environment variables required for ADK os.environ["GOOGLE_GENAI_USE_VERTEXAI"] = "TRUE" os.environ["GOOGLE_CLOUD_PROJECT"] = PROJECT_ID os.environ["GOOGLE_CLOUD_LOCATION"] = LOCATION # Initialize Vertex AI vertexai.init(project=PROJECT_ID, location=LOCATION, staging_bucket=BUCKET_URI)

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}

import logging logging.getLogger("google_adk").setLevel(logging.CRITICAL) logging.getLogger("asyncio").setLevel(logging.CRITICAL) import os import uuid from typing import Any, Iterator, Optional import aiofiles from google.adk.agents import LlmAgent 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 AdkApp

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)

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()

def create_agent(errlog): root_agent = LlmAgent( model="gemini-2.5-flash", name="reddit_assistant_agent", instruction="Help the user fetch reddit info.", tools=[ MCPToolset( connection_params=StdioConnectionParams( server_params=StdioServerParameters( command="mcp-reddit", ), ), errlog=errlog, # Required only in colab environment ) ], ) return root_agent

async def test_agent_locally(): # Create the session service session_service = InMemorySessionService() # Create the specific session where the conversation will happen user_id = "runner_user_01" session = await session_service.create_session( app_name="MyRunnerApp", 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="MyRunnerApp", session_service=session_service ) try: chat_loop(runner, user_id, session.id) finally: # Ensure the log file is always closed await errlog.close()

root_agent_file = f""" import os 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 from vertexai.preview.reasoning_engines import AdkApp # Set variable PROJECT_ID=os.getenv('PROJECT_ID', '{PROJECT_ID}') LOCATION=os.getenv('LOCATION', '{LOCATION}') BUCKET_NAME=os.getenv('BUCKET_NAME', '{BUCKET_NAME}') 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}') # 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=PROJECT_ID, location=LOCATION) agent_app = AdkApp( agent=LlmAgent( model='gemini-2.5-flash', name='reddit_assistant_agent', instruction='Help the user fetch reddit info.', 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()

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", register_operations={ "": ["get_session", "list_sessions", "create_session", "delete_session"], "async": [ "async_get_session", "async_list_sessions", "async_create_session", "async_delete_session", ], "stream": ["stream_query", "streaming_agent_run_with_events"], "async_stream": ["async_stream_query"], }, ), requirements=["google-cloud-aiplatform[agent_engines,adk]>=1.101.0"], extra_packages=[ "root_agent.py", "installation_scripts/install_local_mcp.sh", ], env_vars={ "PROJECT_ID": PROJECT_ID, "LOCATION": LOCATION, "REDDIT_CLIENT_ID": REDDIT_CLIENT_ID, "REDDIT_CLIENT_SECRET": REDDIT_CLIENT_SECRET, "REDDIT_REFRESH_TOKEN": REDDIT_REFRESH_TOKEN, }, build_options={ "installation": [ "installation_scripts/install_local_mcp.sh", ], }, )

delete_agent_engine = True delete_bucket = True if delete_agent_engine: agent_engines = agent_engines.list(filter="display_name=reddit_assistant_agent") for agent_engine in agent_engines: agent_engine.delete(force=True) if delete_bucket: !gsutil rm -r {BUCKET_URI}