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

	Facebook
	Twitter
	Linkedin
	E-Mail

Agentic AI > MCP Protocol > MCP Server with Pub/Sub

MCP Server with Pub/Sub

Author: Venkata Sudhakar

Cloud Pub/Sub is a fully managed messaging service that decouples producers from consumers using topics and subscriptions. ADK agents that need to trigger downstream workflows, broadcast events, or pull messages from queues can benefit from an MCP server that wraps Pub/Sub operations as tools.

In this tutorial, you will build an MCP server with two tools - one to publish a message to a topic and one to pull pending messages from a subscription. An ADK agent uses these tools to send order events and check for incoming notifications.

The MCP server below uses the google-cloud-pubsub library. Messages are base64 encoded by the Pub/Sub API, so the server decodes them before returning to the agent.

# pubsub_server.py
import asyncio
import base64
import json
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
from google.cloud import pubsub_v1

app = Server("pubsub-mcp-server")
PROJECT_ID = "your-project-id"

@app.list_tools()
async def list_tools():
    return [
        Tool(name="pubsub_publish", description="Publish a message to a Pub/Sub topic",
            inputSchema={"type": "object", "properties": {
                "topic_id": {"type": "string"},
                "message": {"type": "string"},
                "attributes": {"type": "object"}
            }, "required": ["topic_id", "message"]}),
        Tool(name="pubsub_pull", description="Pull messages from a Pub/Sub subscription",
            inputSchema={"type": "object", "properties": {
                "subscription_id": {"type": "string"},
                "max_messages": {"type": "integer"}
            }, "required": ["subscription_id"]})
    ]

@app.call_tool()
async def call_tool(name, arguments):
    if name == "pubsub_publish":
        publisher = pubsub_v1.PublisherClient()
        topic_path = publisher.topic_path(PROJECT_ID, arguments["topic_id"])
        data = arguments["message"].encode("utf-8")
        attrs = arguments.get("attributes", {})
        future = publisher.publish(topic_path, data, **attrs)
        msg_id = future.result()
        return [TextContent(type="text", text="Published message ID: " + msg_id)]
    elif name == "pubsub_pull":
        subscriber = pubsub_v1.SubscriberClient()
        sub_path = subscriber.subscription_path(PROJECT_ID, arguments["subscription_id"])
        max_msgs = arguments.get("max_messages", 5)
        response = subscriber.pull(request={"subscription": sub_path, "max_messages": max_msgs})
        messages = []
        ack_ids = []
        for msg in response.received_messages:
            text = base64.b64decode(msg.message.data).decode("utf-8")
            messages.append({"id": msg.message.message_id, "data": text})
            ack_ids.append(msg.ack_id)
        if ack_ids:
            subscriber.acknowledge(request={"subscription": sub_path, "ack_ids": ack_ids})
        return [TextContent(type="text", text=json.dumps(messages))]

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 uses the Pub/Sub MCP server to publish an order event. The agent composes the message payload from the user request and calls the publish tool without the caller managing any Pub/Sub client code directly.

# pubsub_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=["pubsub_server.py"]
        )
    )
    agent = LlmAgent(
        model="gemini-2.0-flash",
        name="pubsub_agent",
        instruction="You handle order events. Publish order messages to Pub/Sub topics and pull notifications from subscriptions as needed.",
        tools=tools
    )
    session_service = InMemorySessionService()
    session = await session_service.create_session(app_name="ps_app", user_id="user1")
    runner = Runner(agent=agent, app_name="ps_app", session_service=session_service)
    response = runner.run(
        user_id="user1",
        session_id=session.id,
        new_message=Content(parts=[Part(text="Publish an order event for order ORD-5521 worth 1200 USD to topic orders-topic")])
    )
    async for event in response:
        if event.is_final_response():
            print(event.content.parts[0].text)
    await exit_stack.aclose()

asyncio.run(main())

This pattern is useful for agents that act as event producers in larger pipelines - for example, triggering fulfilment workflows, sending alerts, or coordinating between microservices. You can extend the server with topic management tools or dead-letter queue monitoring as needed.

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