	In Browser
	StumbleUpon
	del.icio.us
	Google
	Google Buzz
	reddit
	LinkedIn

	Facebook
	Twitter
	Linkedin
	E-Mail

Agentic AI > MCP Protocol > MCP Server with Vertex AI Search

MCP Server with Vertex AI Search

Author: Venkata Sudhakar

Vertex AI Search lets you build enterprise search experiences over structured and unstructured data stored in Google Cloud. ADK agents that need to answer questions from internal knowledge bases, product catalogs, or document repositories can call a Vertex AI Search MCP server to retrieve relevant results without the agent needing to know the underlying index or data schema.

In this tutorial, you will build an MCP server with a search tool that queries a Vertex AI Search data store. An ADK agent then uses this tool to answer user queries by retrieving grounded results from the enterprise search index.

The MCP server below uses the google-cloud-discoveryengine library. Each search call returns a list of result snippets and document titles that the agent can summarise for the user.

# vertexai_search_server.py
import asyncio
import json
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
from google.cloud import discoveryengine_v1beta as discoveryengine

app = Server("vertexai-search-mcp-server")
PROJECT_ID = "your-project-id"
LOCATION = "global"
DATA_STORE_ID = "your-data-store-id"

@app.list_tools()
async def list_tools():
    return [
        Tool(
            name="vertex_search",
            description="Search the enterprise data store using Vertex AI Search",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string"},
                    "page_size": {"type": "integer"}
                },
                "required": ["query"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name, arguments):
    if name == "vertex_search":
        client = discoveryengine.SearchServiceClient()
        serving_config = (
            f"projects/{PROJECT_ID}/locations/{LOCATION}"
            f"/collections/default_collection"
            f"/dataStores/{DATA_STORE_ID}"
            f"/servingConfigs/default_config"
        )
        request = discoveryengine.SearchRequest(
            serving_config=serving_config,
            query=arguments["query"],
            page_size=arguments.get("page_size", 5),
            query_expansion_spec=discoveryengine.SearchRequest.QueryExpansionSpec(
                condition=discoveryengine.SearchRequest.QueryExpansionSpec.Condition.AUTO
            )
        )
        response = client.search(request)
        results = []
        for result in response.results:
            doc = result.document
            title = doc.derived_struct_data.get("title", {}).get("string_value", "Untitled")
            snippet = doc.derived_struct_data.get("snippets", [{}])
            snip_text = snippet[0].get("snippet", {}).get("string_value", "") if snippet else ""
            results.append({"title": title, "snippet": snip_text})
        return [TextContent(type="text", text=json.dumps(results))]

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream, app.create_initialization_options())

if __name__ == "__main__":
    asyncio.run(main())

The ADK agent below connects to the Vertex AI Search MCP server and uses it to answer product-related questions. The agent retrieves grounded results from the data store instead of relying solely on its training knowledge.

# vertexai_search_agent.py
import asyncio
from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, StdioServerParameters
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai.types import Content, Part

async def main():
    tools, exit_stack = await MCPToolset.from_server(
        connection_params=StdioServerParameters(
            command="python", args=["vertexai_search_server.py"]
        )
    )
    agent = LlmAgent(
        model="gemini-2.0-flash",
        name="search_agent",
        instruction="You are an enterprise search assistant. Use the vertex_search tool to find answers from the company knowledge base before responding.",
        tools=tools
    )
    session_service = InMemorySessionService()
    session = await session_service.create_session(app_name="search_app", user_id="user1")
    runner = Runner(agent=agent, app_name="search_app", session_service=session_service)
    response = runner.run(
        user_id="user1",
        session_id=session.id,
        new_message=Content(parts=[Part(text="What is our refund policy for enterprise customers?")])
    )
    async for event in response:
        if event.is_final_response():
            print(event.content.parts[0].text)
    await exit_stack.aclose()

asyncio.run(main())

Vertex AI Search MCP servers are ideal for RAG-style agent workflows where the agent needs to cite company-specific documents rather than generating answers from general knowledge. You can extend the server with a summarise tool, a filter-by-date tool, or a document-fetch tool to build richer retrieval pipelines.

Send your comments, suggestions or queries regarding this site to [email protected].