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

	Facebook
	Twitter
	Linkedin
	E-Mail

Agentic AI > MCP Protocol > MCP Server with Google Cloud Storage

MCP Server with Google Cloud Storage

Author: Venkata Sudhakar

Google Cloud Storage (GCS) is a scalable object storage service that stores files as objects inside buckets. When building AI agents with the Agent Development Kit (ADK), you often need to read reports, write summaries, or manage files stored in GCS. By wrapping GCS operations in an MCP server, your agents get clean, reusable tools for bucket access without embedding storage logic inside the agent itself.

In this tutorial, you will build an MCP server that exposes four GCS tools - list objects, read object, write object, and delete object. An ADK agent then connects to this server and uses these tools to manage files in response to natural language instructions.

The MCP server below uses the google-cloud-storage library to interact with GCS. Each tool maps directly to a GCS API operation and returns results as plain text for the agent to process.

# gcs_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 storage

app = Server("gcs-mcp-server")

@app.list_tools()
async def list_tools():
    return [
        Tool(
            name="gcs_list_objects",
            description="List objects in a GCS bucket",
            inputSchema={
                "type": "object",
                "properties": {
                    "bucket_name": {"type": "string"},
                    "prefix": {"type": "string"}
                },
                "required": ["bucket_name"]
            }
        ),
        Tool(
            name="gcs_read_object",
            description="Read the content of a GCS object",
            inputSchema={
                "type": "object",
                "properties": {
                    "bucket_name": {"type": "string"},
                    "object_name": {"type": "string"}
                },
                "required": ["bucket_name", "object_name"]
            }
        ),
        Tool(
            name="gcs_write_object",
            description="Write content to a GCS object",
            inputSchema={
                "type": "object",
                "properties": {
                    "bucket_name": {"type": "string"},
                    "object_name": {"type": "string"},
                    "content": {"type": "string"}
                },
                "required": ["bucket_name", "object_name", "content"]
            }
        ),
        Tool(
            name="gcs_delete_object",
            description="Delete a GCS object",
            inputSchema={
                "type": "object",
                "properties": {
                    "bucket_name": {"type": "string"},
                    "object_name": {"type": "string"}
                },
                "required": ["bucket_name", "object_name"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name, arguments):
    client = storage.Client()
    if name == "gcs_list_objects":
        prefix = arguments.get("prefix", "")
        blobs = client.list_blobs(arguments["bucket_name"], prefix=prefix)
        objects = [blob.name for blob in blobs]
        return [TextContent(type="text", text=json.dumps(objects))]
    elif name == "gcs_read_object":
        bucket = client.bucket(arguments["bucket_name"])
        blob = bucket.blob(arguments["object_name"])
        content = blob.download_as_text()
        return [TextContent(type="text", text=content)]
    elif name == "gcs_write_object":
        bucket = client.bucket(arguments["bucket_name"])
        blob = bucket.blob(arguments["object_name"])
        blob.upload_from_string(arguments["content"])
        return [TextContent(type="text", text="Written successfully")]
    elif name == "gcs_delete_object":
        bucket = client.bucket(arguments["bucket_name"])
        blob = bucket.blob(arguments["object_name"])
        blob.delete()
        return [TextContent(type="text", text="Deleted successfully")]

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 this MCP server using StdioServerParameters. Once connected, the agent automatically discovers the GCS tools and can use them to answer storage-related queries from users.

# gcs_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=["gcs_server.py"]
        )
    )
    agent = LlmAgent(
        model="gemini-2.0-flash",
        name="gcs_agent",
        instruction="You are a GCS file manager. Help users manage Cloud Storage buckets.",
        tools=tools
    )
    session_service = InMemorySessionService()
    session = await session_service.create_session(app_name="gcs_app", user_id="user1")
    runner = Runner(agent=agent, app_name="gcs_app", session_service=session_service)
    response = runner.run(
        user_id="user1",
        session_id=session.id,
        new_message=Content(parts=[Part(text="List all objects in bucket my-data-bucket")])
    )
    async for event in response:
        if event.is_final_response():
            print(event.content.parts[0].text)
    await exit_stack.aclose()

asyncio.run(main())

The GCS MCP server pattern works for any bucket operation. You can extend it with additional tools such as copying objects between buckets, generating signed URLs, or reading object metadata. This keeps your agent focused on business logic while the MCP server handles all storage concerns.

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