Real-World Agent Project: MySQL & Streaming

This final project connects an agent to a real production database. The public Olist e-commerce dataset goes into a cloud MySQL server. An agent then reaches it through a read-only MySQL MCP server. The agent answers natural-language questions by writing and running SQL, but it cannot change our data, because writes are disabled at the connection level.

In this blog, the last step serves it as a streaming FastAPI assistant, reusing the deployment pattern from the previous lesson.

Note

This lesson builds on the streaming server from Deploy AI Agents with FastAPI and the mcp_config.json pattern from the MCP projects.

The Dataset

We use the Brazilian e-commerce dataset by Olist, distributed as a single SQLite file. It contains customers, orders, order items, payments, reviews, products, sellers, geolocation, and marketing leads.

Tip

Find the dataset by searching "Olist e-commerce dataset SQLite Kaggle". Download the .sqlite file and place it at db/olist.sqlite in your project.

Get a Free Cloud MySQL Database

We need a MySQL server reachable from our code. TiDB Cloud offers a free, MySQL-compatible serverless tier that works well for this project.

Create a free account (search "TiDB Cloud pingcap.com" for the official site).
Create a serverless cluster and open its Connect dialog.
Copy the host, port, user, and password, and note that SSL is required.

Add the connection details to your .env file:

BASH

MYSQL_HOST=your-tidb-host.tidbcloud.com
MYSQL_PORT=4000
MYSQL_USER=your-username
MYSQL_PASSWORD=your-password

Important

Never commit real database credentials to source control. Keep them in .env (which should be in .gitignore) and reference them with os.getenv(...).

Migrating SQLite to MySQL

A short notebook copies every table from the local SQLite file into the cloud MySQL database. Import the drivers and load environment variables:

Copy every SQLite table into a cloud MySQL ecommerce database

PYTHON

import warnings
warnings.filterwarnings('ignore')

import os
from dotenv import load_dotenv
load_dotenv()

import sqlite3, pymysql

Open both connections. The MySQL connection reads its settings from .env and enables SSL (required by TiDB Cloud):

PYTHON

sqlite_conn = sqlite3.connect('db/olist.sqlite')
sqlite_cur = sqlite_conn.cursor()

mysql_conn = pymysql.connect(
    host=os.getenv("MYSQL_HOST"),
    port=int(os.getenv('MYSQL_PORT')),
    user=os.getenv("MYSQL_USER"),
    password=os.getenv("MYSQL_PASSWORD"),
    ssl={'ssl': {}}
)
mysql_cur = mysql_conn.cursor()

Create the target database and select it:

PYTHON

mysql_cur.execute("CREATE DATABASE IF NOT EXISTS ecommerce")
mysql_cur.execute("USE ecommerce")

Now loop over every SQLite table, recreate it in MySQL (all columns as TEXT for a simple, type-safe copy), and bulk-insert the rows:

PYTHON

# Get all tables from SQLite
sqlite_cur.execute("SELECT name FROM sqlite_master WHERE type='table'")
tables = [row[0] for row in sqlite_cur.fetchall()]

for table in tables:
    print(f"Copying table: {table}")

    # Recreate the table in MySQL
    sqlite_cur.execute(f"PRAGMA table_info({table})")
    columns = [f"{col[1]} TEXT" for col in sqlite_cur.fetchall()]
    mysql_cur.execute(f"CREATE TABLE IF NOT EXISTS {table} ({', '.join(columns)})")

    # Copy the data
    sqlite_cur.execute(f"SELECT * FROM {table}")
    rows = sqlite_cur.fetchall()
    if rows:
        placeholders = ", ".join(["%s"] * len(rows[0]))
        mysql_cur.executemany(f"INSERT INTO {table} VALUES ({placeholders})", rows)

    mysql_conn.commit()

print("Done!")
sqlite_conn.close()
mysql_conn.close()

OUTPUT

Copying table: product_category_name_translation
Copying table: sellers
Copying table: customers
Copying table: geolocation
Copying table: order_items
Copying table: order_payments
Copying table: order_reviews
Copying table: orders
Copying table: products
Copying table: leads_qualified
Copying table: leads_closed
Done!

Our cloud MySQL ecommerce database now mirrors the SQLite dataset.

Connecting the Agent with a Read-Only MySQL Server

The agent talks to MySQL through the @benborla29/mcp-server-mysql server. The key part is the configuration: all write operations are disabled, so the agent can only read. Add this entry to scripts/mcp_config.json:

Insert, update, and delete are disabled, so the agent only reads

JSON

{
  "tidb_ecommerce": {
    "command": "npx",
    "args": ["-y", "@benborla29/mcp-server-mysql"],
    "env": {
      "MYSQL_HOST": "your-tidb-host.tidbcloud.com",
      "MYSQL_PORT": "4000",
      "MYSQL_USER": "your-username",
      "MYSQL_PASS": "your-password",
      "MYSQL_DB": "ecommerce",
      "ALLOW_INSERT_OPERATION": "false",
      "ALLOW_UPDATE_OPERATION": "false",
      "ALLOW_DELETE_OPERATION": "false",
      "MYSQL_SSL": "true",
      "MYSQL_DISABLE_READ_ONLY_TRANSACTIONS": "true"
    },
    "transport": "stdio"
  }
}

Caution

Replace the placeholder host, user, and password with your own TiDB Cloud values. Never paste real credentials into a file you might commit. Keeping ALLOW_INSERT_OPERATION, ALLOW_UPDATE_OPERATION, and ALLOW_DELETE_OPERATION set to false is what makes this agent safe: even if the model writes a destructive query, the server refuses to run it.

Note

Search "benborla29 mcp-server-mysql GitHub" for the server's full configuration options and the list of tools it exposes.

The Streaming MySQL Assistant

The deployment reuses the streaming FastAPI pattern, with a system prompt that scopes the agent to database questions only. Tools are loaded once at startup from the single tidb_ecommerce server:

A question becomes SQL, runs on MySQL, and streams the answer

PYTHON

import sys
import os

root_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.append(root_dir)

from dotenv import load_dotenv
load_dotenv()

from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
import json

from langchain_google_genai import ChatGoogleGenerativeAI
from langchain.agents import create_agent
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.checkpoint.memory import InMemorySaver
from langchain.messages import HumanMessage, AIMessageChunk

from scripts import utils

checkpointer = InMemorySaver()
tools = None

class ChatRequest(BaseModel):
    query: str = Field(..., min_length=2)
    model: str = "gemini-2.5-flash"
    thread_id: str = "default"

async def get_tools():
    mcp_config = utils.load_mcp_config("tidb_ecommerce")
    client = MultiServerMCPClient(mcp_config)
    safe_tools = await client.get_tools()

    print(f"Loaded {len(safe_tools)} Tools")
    print(f"Tools Available\n{[tool.name for tool in safe_tools]}")
    return safe_tools

@asynccontextmanager
async def lifespan(app: FastAPI):
    global tools
    tools = await get_tools()
    print("Tools are loaded. Ready to create agent!")
    yield

app = FastAPI(lifespan=lifespan)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"], allow_methods=["*"], allow_headers=["*"],
)

The streaming generator builds the agent with a database-scoped prompt and streams its output as newline-delimited JSON:

PYTHON

async def stream_response(query, model_name, thread_id):
    system_prompt = """You are a MySQL Assistant agent.
                    You have access to a MySQL server. Answer user queries by accessing data
                    from the MySQL server. If a query is not related to the database, tell the
                    user to ask database-related questions only."""

    model = ChatGoogleGenerativeAI(model=model_name)
    agent = create_agent(
        model=model, tools=tools, system_prompt=system_prompt, checkpointer=checkpointer
    )

    config = {"configurable": {"thread_id": thread_id}}

    async for chunk, metadata in agent.astream(
            {"messages": [HumanMessage(query)]}, stream_mode="messages", config=config):

        data = {"type": chunk.__class__.__name__, "content": chunk.text}
        if isinstance(chunk, AIMessageChunk) and chunk.tool_calls:
            data["tool_calls"] = chunk.tool_calls

        yield (json.dumps(data) + "\n").encode()

Expose the endpoints exactly as in the previous lesson:

PYTHON

@app.get("/")
async def read_root():
    return {"status": "MySQL assistant server is up!"}

@app.post("/chat_stream")
async def chat_stream(request: ChatRequest):
    if not request.query.strip():
        raise HTTPException(status_code=400, detail="Empty prompt!")
    try:
        return StreamingResponse(
            stream_response(request.query, request.model, request.thread_id),
            media_type="application/x-ndjson")
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Server error: {e}")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app=app, host="0.0.0.0", port=8002)

Running the Assistant

Start the server:

POWERSHELL

python "04 Real-World Projects/04_02_stream_server.py"

Then point the Streamlit client from the previous lesson at http://localhost:8002/chat_stream, or test directly with curl:

BASH

curl -X POST http://localhost:8002/chat_stream \
  -H "Content-Type: application/json" \
  -d "{\"query\": \"How many orders were delivered in 2018?\", \"thread_id\": \"analyst-1\"}"

OUTPUT

{"type": "AIMessageChunk", "tool_calls": [{"name": "mysql_query", "args": {"sql": "SELECT COUNT(*) FROM orders WHERE order_status='delivered' AND order_delivered_customer_date LIKE '2018%'"}, ...}]}
{"type": "AIMessageChunk", "content": "In 2018, "}
{"type": "AIMessageChunk", "content": "there were 51,255 delivered orders."}

Here, we can see the agent translate the question into SQL, run it against cloud MySQL through MCP, and stream the answer. Try richer analytics:

PLAINTEXT

You: What are the top 5 product categories by total revenue?
You: Which states have the most customers?
You: What's the average review score per payment type?

Each one is answered by generated SQL over the real ecommerce database. And because writes are disabled, a malformed or malicious query can never alter our data.

Tip

This "text-to-SQL over MCP" pattern generalizes to any MySQL database. Point the server at your own schema, keep the write flags false for analytics use cases, and you have a safe natural-language interface to production data.

Series Wrap-Up

Across this series, we went from a first Gemini call to a deployed, read-only database agent:

Set up Gemini 3, LangChain & LangSmith.
Mastered agent fundamentals, tools, memory, middleware, guardrails, and prompts.
Connected real services through MCP: hotel search, travel planning, code execution, spreadsheets, and a daily briefing.
Deployed agents over HTTP with FastAPI and Streamlit.
Built this real-world, read-only MySQL streaming assistant.

This is how we ship a production database agent. We now have the full toolkit to design, secure, and deploy production AI agents.

Real-World Agent Project: MySQL & Streaming

The Dataset

Get a Free Cloud MySQL Database

Migrating SQLite to MySQL

Connecting the Agent with a Read-Only MySQL Server

The Streaming MySQL Assistant

Running the Assistant

Series Wrap-Up

Found this useful? Keep building with me.

Latest recommendations you might like

Deploy AI Agents with FastAPI

Build a Daily Briefing AI Agent

Build a Google Sheets Analysis Agent with MCP

Build a Code Execution Agent with E2B

Find this tutorial useful?

Discussion & Comments