Integrări MCP cu baze de date

Introducere în Model Context Protocol (MCP)

James Chapman

AI Curriculum Manager, DataCamp

De ce să adăugăm o bază de date la serverul de fuse orare?

Putere de interogare: filtrare, căutare și sortare fără a încărca întregul set de date
Scalabilitate: căutări rapide cu indexuri de baze de date

Conexiune la baza de date

De ce să adăugăm o bază de date la serverul de fuse orare?

Putere de interogare: filtrare, căutare și sortare fără a încărca întregul set de date
Scalabilitate: căutări rapide cu indexuri de baze de date
Concurență: apeluri multiple de instrumente sau clienți accesează BD în siguranță
Pregătit pentru producție: backup-uri, replicare și o sursă unică de adevăr

Avantaje baze de date

Acces la date: Instrumente vs. Resurse

Pictogramă instrumente

Dinamice, dependente de intrările utilizatorului, operații de scriere
Exemple: Analize și operații pe date

Pictogramă resurse

Doar citire, date statice, informații de referință
Exemple: Ghiduri, documentație, politici

Ciclul de viață al conexiunii

Conexiuni la baze de date

Resursă bază de date: timezone_server.py

from mcp.server.fastmcp import FastMCP
import sqlite3

mcp = FastMCP("Timezone Converter")

# Create connection at startup; reuse in handlers
conn = sqlite3.connect("timezones.db")
conn.row_factory = sqlite3.Row

@mcp.resource("db://timezones")
def get_locations() -> str:
    try:
        cursor = conn.execute("SELECT timezone FROM locations")
        rows = cursor.fetchall()
        return "\n".join(r["timezone"] for r in rows)
    except sqlite3.Error as e: return f"Error: {e}"

Instrument de căutare în BD: timezone_server.py

@mcp.tool()
def lookup_locations(prefix: str) -> str:
    """Find timezones that contain the given prefix."""
    try:
        cursor = conn.execute(
            "SELECT timezone FROM locations WHERE timezone LIKE ? LIMIT 50", (f"%{prefix}%",))
        rows = cursor.fetchall()
        return "\n".join(r["timezone"] for r in rows)
    except sqlite3.Error as e: return f"Error: {e}"


if __name__ == "__main__":
    try:
        mcp.run(transport="stdio")
    finally:
        conn.close()

Bune practici de siguranță și producție

Interogări parametrizate (?): nu formatați niciodată ca șir intrările utilizatorului sau ale LLM în SQL
- Injectare prin prompt → injectare malițioasă de cod prin prompturi

SELECT timezone FROM locations WHERE timezone LIKE ? LIMIT 50

Preferați acces la baza de date doar în citire și cu domeniu restrâns pentru serverul MCP
Aplicați limite: număr maxim de rânduri și timeout-uri

Să exersăm!

Introducere în Model Context Protocol (MCP)