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

	Facebook
	Twitter
	Linkedin
	E-Mail

Agentic AI > MCP Protocol > MCP Server with Cloud Spanner

MCP Server with Cloud Spanner

Author: Venkata Sudhakar

Cloud Spanner is Google's fully managed, globally distributed relational database that offers strong consistency and high availability. By wrapping Spanner as an MCP server, AI agents can run SQL queries, insert records, and perform transactional operations through simple tool calls.

ShopMax India uses Cloud Spanner as its inventory and orders database across Mumbai, Bangalore, and Delhi regions. An MCP server exposes Spanner operations as tools so that ADK agents can query stock levels, process orders, and check regional availability without any direct database credentials.

The below example shows an MCP server that exposes two tools - one for querying product inventory and another for inserting an order record into Cloud Spanner.

# Install: pip install mcp google-cloud-spanner
import asyncio
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types
from google.cloud import spanner

# Spanner config
PROJECT_ID = "shopmax-india"
INSTANCE_ID = "shopmax-main"
DATABASE_ID = "inventory"

client = spanner.Client(project=PROJECT_ID)
instance = client.instance(INSTANCE_ID)
database = instance.database(DATABASE_ID)

app = Server("spanner-mcp")

@app.list_tools()
async def list_tools():
    return [
        types.Tool(
            name="query_inventory",
            description="Query product inventory from Cloud Spanner",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "City name: Mumbai, Bangalore, Delhi"},
                    "category": {"type": "string", "description": "Product category e.g. Laptops"}
                },
                "required": ["city"]
            }
        ),
        types.Tool(
            name="insert_order",
            description="Insert a new order record into Cloud Spanner",
            inputSchema={
                "type": "object",
                "properties": {
                    "order_id": {"type": "string"},
                    "product_id": {"type": "string"},
                    "city": {"type": "string"},
                    "quantity": {"type": "integer"},
                    "amount": {"type": "number"}
                },
                "required": ["order_id", "product_id", "city", "quantity", "amount"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "query_inventory":
        city = arguments["city"]
        category = arguments.get("category", "")
        sql = "SELECT product_id, product_name, stock_qty, price_rs FROM inventory WHERE city = @city"
        params = {"city": city}
        param_types = {"city": spanner.param_types.STRING}
        if category:
            sql += " AND category = @category"
            params["category"] = category
            param_types["category"] = spanner.param_types.STRING
        rows = []
        with database.snapshot() as snapshot:
            results = snapshot.execute_sql(sql, params=params, param_types=param_types)
            for row in results:
                rows.append({"product_id": row[0], "name": row[1], "stock": row[2], "price_rs": row[3]})
        return [types.TextContent(type="text", text=str(rows))]

elif name == "insert_order":
        def insert_tx(transaction):
            transaction.insert(
                "orders",
                columns=["order_id", "product_id", "city", "quantity", "amount_rs", "created_at"],
                values=[
                    (arguments["order_id"], arguments["product_id"], arguments["city"],
                     arguments["quantity"], arguments["amount"],
                     spanner.COMMIT_TIMESTAMP)
                ]
            )
        database.run_in_transaction(insert_tx)
        return [types.TextContent(type="text", text="Order inserted 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: "Show Laptops stock in Mumbai"
Tool: query_inventory({city: "Mumbai", category: "Laptops"})

[{"product_id": "LAP001", "name": "Dell Inspiron 15", "stock": 42, "price_rs": 55000},
 {"product_id": "LAP002", "name": "HP Pavilion 14", "stock": 18, "price_rs": 49000},
 {"product_id": "LAP003", "name": "Lenovo IdeaPad 3", "stock": 67, "price_rs": 43000}]

# Agent query: "Create order ORD-9001 for LAP001 in Mumbai, qty 2 at Rs 110000"
Tool: insert_order({order_id: "ORD-9001", product_id: "LAP001", city: "Mumbai", quantity: 2, amount: 110000})

Order inserted successfully

For production deployments, use Spanner's read-only replicas for query tools to reduce costs, while reserving read-write transactions for mutation tools. Apply IAM roles at the database level so the MCP server service account has only the permissions it needs. Use Spanner's built-in query statistics to monitor slow queries from agent workloads.

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