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

	Facebook
	Twitter
	Linkedin
	E-Mail

Agentic AI > MCP Protocol > MCP Server for Database Access

MCP Server for Database Access

Author: Venkata Sudhakar

Putting database access logic inside an ADK agent creates tight coupling - every agent that needs data must duplicate the connection and query code. A better pattern is to encapsulate all database access inside an MCP server. The server owns the connection pool, enforces parameterised queries to prevent SQL injection, and exposes clean tool interfaces. Any ADK agent connects to this server and gets database capabilities without knowing any SQL internals.

FastMCP makes it straightforward to wrap SQLite or PostgreSQL queries as MCP tools. Each tool accepts typed parameters, runs a safe parameterised query, and returns structured results. The MCP server runs as a subprocess started by MCPToolset, so it shares the same process lifetime as the agent session.

The below example shows a ShopMax India MCP server exposing order database queries to an ADK agent.

# shopmax_db_server.py - MCP server with database tools
import sqlite3
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("shopmax-db")

def get_connection():
    conn = sqlite3.connect(":memory:")
    conn.row_factory = sqlite3.Row
    # Seed demo data
    conn.execute("""
        CREATE TABLE orders (
            order_id TEXT, customer TEXT, city TEXT,
            product TEXT, amount_rs INTEGER, status TEXT
        )
    """)
    conn.executemany("INSERT INTO orders VALUES (?,?,?,?,?,?)", [
        ("ORD-001","Priya","Mumbai","Laptop",45000,"Delivered"),
        ("ORD-002","Arjun","Delhi","TV",35000,"Shipped"),
        ("ORD-003","Meera","Bangalore","Phone",18000,"Delivered"),
        ("ORD-004","Rohit","Hyderabad","Laptop",45000,"Processing"),
        ("ORD-005","Sneha","Chennai","TV",35000,"Delivered"),
    ])
    conn.commit()
    return conn

DB = get_connection()

@mcp.tool()
def get_orders_by_city(city: str) -> list:
    # Fetch all orders for a given city
    rows = DB.execute(
        "SELECT order_id, customer, product, amount_rs, status FROM orders WHERE city = ?",
        (city,)
    ).fetchall()
    return [dict(r) for r in rows]

@mcp.tool()
def get_revenue_summary() -> dict:
    # Aggregate total revenue and order count by status
    rows = DB.execute(
        "SELECT status, COUNT(*) as count, SUM(amount_rs) as total_rs FROM orders GROUP BY status"
    ).fetchall()
    return {"summary": [dict(r) for r in rows]}

@mcp.tool()
def get_order_by_id(order_id: str) -> dict:
    # Fetch a single order by ID
    row = DB.execute(
        "SELECT * FROM orders WHERE order_id = ?", (order_id,)
    ).fetchone()
    return dict(row) if row else {"error": "Order not found"}

if __name__ == "__main__":
    mcp.run(transport="stdio")

Now connect an ADK agent to query the database via MCP,

# adk_db_agent.py
import asyncio
from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, StdioServerParameters
from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner
from google.genai import types

async def main():
    mcp_tools, exit_stack = await MCPToolset.from_server(
        connection_params=StdioServerParameters(
            command="python", args=["shopmax_db_server.py"]
        )
    )

agent = LlmAgent(
        name="db_agent",
        model="gemini-2.0-flash",
        tools=mcp_tools,
        instruction="You are a ShopMax India data analyst. Use database tools to answer business questions."
    )

session_service = InMemorySessionService()
    runner = Runner(agent=agent, app_name="db_agent", session_service=session_service)
    session = session_service.create_session(app_name="db_agent", user_id="analyst1")

queries = [
        "What orders do we have in Mumbai?",
        "Give me a revenue summary by order status.",
    ]
    for q in queries:
        msg = types.Content(role="user", parts=[types.Part(text=q)])
        async for event in runner.run_async(user_id="analyst1", session_id=session.id, new_message=msg):
            if event.is_final_response():
                print("Q:", q)
                print("A:", event.content.parts[0].text)
                print()

await exit_stack.aclose()

asyncio.run(main())

It gives the following output,

Q: What orders do we have in Mumbai?
A: ShopMax India has 1 order from Mumbai:
- ORD-001: Priya ordered a Laptop for Rs 45,000. Status: Delivered.

Q: Give me a revenue summary by order status.
A: Here is the revenue summary:
- Delivered: 3 orders, total Rs 98,000
- Shipped: 1 order, total Rs 35,000
- Processing: 1 order, total Rs 45,000

The MCP server enforces parameterised queries throughout, making SQL injection impossible regardless of what the agent passes. In production, replace the in-memory SQLite database with a PostgreSQL connection using psycopg2 or asyncpg. Use a connection pool (e.g. psycopg2.pool.ThreadedConnectionPool) so the MCP server can handle concurrent tool calls from multiple agent sessions. Add read-only database credentials to the MCP server so agents cannot accidentally mutate data through the query tools.

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