Integracje MCP z bazami danych

Wprowadzenie do Model Context Protocol (MCP)

James Chapman

AI Curriculum Manager, DataCamp

Dlaczego dodać bazę danych do serwera stref czasowych?

Możliwości zapytań: filtrowanie, wyszukiwanie i sortowanie bez ładowania całego zbioru danych
Skalowalność: szybkie wyszukiwania dzięki indeksom bazy danych

Połączenie z bazą danych

Dlaczego dodać bazę danych do serwera stref czasowych?

Możliwości zapytań: filtrowanie, wyszukiwanie i sortowanie bez ładowania całego zbioru danych
Skalowalność: szybkie wyszukiwania dzięki indeksom bazy danych
Współbieżność: wiele wywołań narzędzi lub klientów bezpiecznie korzysta z bazy
Gotowość produkcyjna: kopie zapasowe, replikacja i jedno źródło danych

Zalety bazy danych

Dostęp do danych: narzędzia a zasoby

Ikona narzędzi

Dynamiczne operacje, silnie zależne od danych użytkownika, w tym operacje zapisu
Przykłady: Analityka i operacje na danych

Ikona zasobów

Tylko do odczytu, dane statyczne, informacje referencyjne
Przykłady: Wytyczne, dokumentacja, zasady

Cykl życia połączenia

Cykl życia połączenia z bazą danych

Zasób bazy danych: 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}"

Narzędzie wyszukiwania w bazie: 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()

Bezpieczeństwo i dobre praktyki produkcyjne

Zapytania parametryzowane (?): nigdy nie formatuj bezpośrednio danych użytkownika ani LLM w SQL
- Prompt injection → złośliwe wstrzykiwanie kodu przez prompty

SELECT timezone FROM locations WHERE timezone LIKE ? LIMIT 50

Preferuj dostęp tylko do odczytu i ściśle ograniczony dostęp do bazy dla serwera MCP
Stosuj limity: maksymalna liczba wierszy i limity czasu

Czas na ćwiczenia!

Wprowadzenie do Model Context Protocol (MCP)