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

	Facebook
	Twitter
	Linkedin
	E-Mail

Agentic AI > MCP Protocol > MCP Server with Cloud SQL

MCP Server with Cloud SQL

Author: Venkata Sudhakar

Cloud SQL is Google Cloud's fully managed relational database service supporting MySQL, PostgreSQL, and SQL Server. For teams already running Cloud SQL in production, building an MCP server on top of it is the fastest way to give AI agents structured data access without changing the database layer.

ShopMax India uses Cloud SQL MySQL for its product catalogue and pricing engine. The MCP server below uses the Cloud SQL Python Connector for secure IAM-authenticated connections, and exposes tools to fetch product details and update prices - operations the pricing agent uses daily.

The below example shows an MCP server connecting to Cloud SQL MySQL using the google-cloud-sql-connector library and exposing product catalogue tools.

# Install: pip install mcp cloud-sql-python-connector[pymysql] sqlalchemy
import asyncio
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types
from google.cloud.sql.connector import Connector
import sqlalchemy

INSTANCE_CONNECTION_NAME = "shopmax-india:asia-south1:shopmax-mysql"
DB_USER = "shopmax_agent"
DB_PASS = "secret"
DB_NAME = "catalogue"

connector = Connector()

def get_conn():
    return connector.connect(
        INSTANCE_CONNECTION_NAME,
        "pymysql",
        user=DB_USER,
        password=DB_PASS,
        db=DB_NAME
    )

engine = sqlalchemy.create_engine("mysql+pymysql://", creator=get_conn)

app = Server("cloudsql-mcp")

@app.list_tools()
async def list_tools():
    return [
        types.Tool(
            name="get_product",
            description="Fetch product details from the catalogue by product ID or name",
            inputSchema={
                "type": "object",
                "properties": {
                    "identifier": {"type": "string", "description": "Product ID or partial name"}
                },
                "required": ["identifier"]
            }
        ),
        types.Tool(
            name="update_price",
            description="Update the price of a product in the catalogue",
            inputSchema={
                "type": "object",
                "properties": {
                    "product_id": {"type": "string"},
                    "new_price_rs": {"type": "number"}
                },
                "required": ["product_id", "new_price_rs"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    with engine.connect() as conn:
        if name == "get_product":
            identifier = arguments["identifier"]
            result = conn.execute(
                sqlalchemy.text(
                    "SELECT product_id, name, category, price_rs, stock "
                    "FROM products WHERE product_id = :id OR name LIKE :name LIMIT 5"
                ),
                {"id": identifier, "name": "%" + identifier + "%"}
            )
            rows = [dict(r._mapping) for r in result]
            return [types.TextContent(type="text", text=str(rows))]

elif name == "update_price":
            conn.execute(
                sqlalchemy.text(
                    "UPDATE products SET price_rs = :price, updated_at = NOW() "
                    "WHERE product_id = :pid"
                ),
                {"price": arguments["new_price_rs"], "pid": arguments["product_id"]}
            )
            conn.commit()
            return [types.TextContent(type="text", text="Price updated 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: "Get details for Samsung TV"
Tool: get_product({identifier: "Samsung TV"})

[{"product_id": "TV-201", "name": "Samsung 55 Crystal 4K", "category": "Televisions",
  "price_rs": 62000, "stock": 15},
 {"product_id": "TV-202", "name": "Samsung 43 Smart LED", "category": "Televisions",
  "price_rs": 38000, "stock": 28}]

# Agent query: "Update price of TV-201 to Rs 59000"
Tool: update_price({product_id: "TV-201", new_price_rs: 59000})

Price updated successfully

For production deployments, use IAM database authentication instead of password-based credentials so the MCP server service account authenticates directly without storing passwords. Enable Cloud SQL read replicas and direct read-only MCP tools to the replica endpoint to reduce load on the primary. Use connection pooling with a max pool size tuned to the Cloud SQL instance tier to avoid connection exhaustion under concurrent agent load.

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