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

	Facebook
	Twitter
	Linkedin
	E-Mail

Agentic AI > MCP Protocol > MCP Server with AlloyDB

MCP Server with AlloyDB

Author: Venkata Sudhakar

AlloyDB is Google Cloud's fully managed PostgreSQL-compatible database designed for demanding enterprise workloads. It delivers superior performance for both transactional and analytical queries. Wrapping AlloyDB as an MCP server lets AI agents perform natural language queries, lookups, and writes against a production-grade relational database.

ShopMax India runs its customer and returns management system on AlloyDB. An MCP server built on top of AlloyDB gives ADK agents the ability to search customer records, check return status, and update case notes - all through standard MCP tool calls using a connection pool for efficiency.

The below example shows an MCP server connecting to AlloyDB via asyncpg and exposing tools to search customers and log return requests.

# Install: pip install mcp asyncpg
import asyncio
import asyncpg
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types

# AlloyDB connection string (use Cloud SQL Auth Proxy or PSC endpoint)
DSN = "postgresql://shopmax_agent:secret@127.0.0.1:5432/shopmax"

app = Server("alloydb-mcp")
pool = None

async def get_pool():
    global pool
    if pool is None:
        pool = await asyncpg.create_pool(DSN, min_size=2, max_size=10)
    return pool

@app.list_tools()
async def list_tools():
    return [
        types.Tool(
            name="search_customer",
            description="Search customer records by name or phone number",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Name or phone number to search"}
                },
                "required": ["query"]
            }
        ),
        types.Tool(
            name="log_return_request",
            description="Log a product return request for a customer order",
            inputSchema={
                "type": "object",
                "properties": {
                    "order_id": {"type": "string"},
                    "reason": {"type": "string"},
                    "refund_amount_rs": {"type": "number"}
                },
                "required": ["order_id", "reason", "refund_amount_rs"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    db = await get_pool()
    if name == "search_customer":
        q = "%" + arguments["query"] + "%"
        rows = await db.fetch(
            "SELECT customer_id, name, phone, city FROM customers "
            "WHERE name ILIKE $1 OR phone LIKE $1 LIMIT 10",
            q
        )
        results = [dict(r) for r in rows]
        return [types.TextContent(type="text", text=str(results))]

elif name == "log_return_request":
        await db.execute(
            "INSERT INTO return_requests (order_id, reason, refund_amount_rs, status, created_at) "
            "VALUES ($1, $2, $3, $4, NOW())",
            arguments["order_id"], arguments["reason"],
            arguments["refund_amount_rs"], "PENDING"
        )
        return [types.TextContent(type="text", text="Return request logged successfully")]

async def main():
    async with stdio_server() as streams:
        await app.run(streams[0], streams[1], app.create_initialization_options())

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

It gives the following output,

# Agent query: "Find customer Priya Sharma"
Tool: search_customer({query: "Priya Sharma"})

[{"customer_id": "C-4421", "name": "Priya Sharma", "phone": "9876543210", "city": "Bangalore"},
 {"customer_id": "C-4892", "name": "Priya Sharma", "phone": "9812345678", "city": "Mumbai"}]

# Agent query: "Log return for order ORD-8812, defective product, refund Rs 45000"
Tool: log_return_request({order_id: "ORD-8812", reason: "Defective product", refund_amount_rs: 45000})

Return request logged successfully

For production use, connect AlloyDB using the AlloyDB Auth Proxy instead of exposing the database endpoint directly. Use connection pooling with PgBouncer or asyncpg pool settings tuned for agent concurrency. AlloyDB's columnar engine accelerates analytical queries, so bulk reporting tools in the MCP server benefit from enabling it on the relevant tables.

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