Stwórz serwer MCP w Pythonie i połącz go z Claude Desktop oraz SQLite.

Każdy, kto tworzy agenty korporacyjne, zna trudności związane z pisaniem kodu łączącego różne komponenty. Jeden wrapper do bazy danych, kolejna warstwa integracji dla zleceń w Jira, jeszcze inna dla Slacka. Pomnóż to przez dziesięć baz danych i pięć frameworki LLM takich jak Claude, AutoGen i LangChain, a projekt załamie się pod własnym ciężarem. To właśnie jest problem integracji typu N przez M: sztuczna inteligencja musi rozumieć dokładną strukturę każdego źródła danych, z którym ma do czynienia.

Model Context Protocol (MCP) to otwarty standard skierowany bezpośrednio na ten problem. Wyobraź sobie uniwersalny kabel USB-C dla sztucznej inteligencji. Połączenie z bazą danych jest tworzone raz i udostępniane przez pojedynczy, standaryzowany serwer, dzięki czemu każdy kompatybilny klient może się do niego połączyć, odkryć dostępne narzędzia i je używać. Poniżej przedstawiono lokalny serwer MCP składający się z mniej niż 50 linijek kodu w Pythonie, połączony z Claude Desktop i wspierany przez żywą bazę danych SQLite.

Wymagania wstępne

Oficjalny pakiet Python mcp zajmuje się tym trudnym zadaniem. Ukrywa szczegóły JSON-RPC, dzięki czemu wystarczy jeden dekorator, aby udostępnić narzędzie modelowi. Najpierw zainstaluj go w swoim środowisku wirtualnym:

pip install mcp

Claude Desktop musi również zostać zainstalowany na tym komputerze.

Tworzenie serwera

Stwórz plik o nazwie mcp_server.py, importuj niezbędne biblioteki i uruchom serwer. Ważny jest jeden szczegół: moduł os umieszcza plik SQLite w tym samym katalogu co skrypt. Claude Desktop uruchamia serwer w tle, a bez określonego miejsca przechowywania mogą pojawić się błędy uprawnień.

import sqlite3
import json
import os
import sys
from mcp.server.mcpserver import MCPServer

# Initialize the MCP server
mcp = MCPServer(name="Enterprise_SQL_Agent")

# Force the database to be created in the exact same folder as this script
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
DB_PATH = os.path.join(BASE_DIR, "enterprise.db")

Następnie tworzy się tymczasowa baza danych, aby model miał coś do zapytania. Rejestracja działań odbywa się do standardowego błędu (sys.stderr). Standardowy wyjście to kanał, którego używa MCP do komunikacji z klientem, a przypadkowe wyświetlanie treści tam mogłoby go uszkodzić.

def setup_dummy_db():
    """Create a sample employee database for the demo"""
    try:
        conn = sqlite3.connect(DB_PATH)
        cursor = conn.cursor()

        cursor.execute('''CREATE TABLE IF NOT EXISTS employees
                          (id INTEGER PRIMARY KEY, name TEXT, role TEXT, salary INTEGER)''')
        cursor.execute("DELETE FROM employees")

        employees = [
            ("Alice", "Data Scientist", 120000),
            ("Bob", "DevOps Engineer", 115000),
            ("Charlie", "AI Researcher", 135000)
        ]

        cursor.executemany("INSERT INTO employees (name, role, salary) VALUES (?, ?, ?)", employees)
        conn.commit()
        conn.close()
        print("Database initialized successfully.", file=sys.stderr)
    except Exception as e:
        print(f"Database setup error: {e}", file=sys.stderr)

Używanie bazy danych jako narzędzia

Głównym elementem jest zwykła funkcja w Pythonie, która wykonywać zapytanie SQL i zwraca wyniki. Dodając dekorator @mcp.tool(), serwer odczytuje dokumentację i wskazówki typowe, aby automatycznie stworzyć schemat JSON dla modelu.

@mcp.tool()
def query_employee_database(sql_query: str) -> str:
    """
    Executes a SQL SELECT query against the enterprise.db database.

    The database contains an 'employees' table with columns:
    - id (INTEGER PRIMARY KEY)
    - name (TEXT)
    - role (TEXT)
    - salary (INTEGER)

    SECURITY: Only READ operations (SELECT) are permitted.
    """

    # Safety Check: Block destructive SQL commands
    dangerous_keywords = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER"]
    if any(keyword in sql_query.upper() for keyword in dangerous_keywords):
        return "Error: Only SELECT queries are authorized for this tool."

    try:
        conn = sqlite3.connect(DB_PATH)
        cursor = conn.cursor()
        cursor.execute(sql_query)
        results = cursor.fetchall()

        # Format the output as JSON so the LLM can read it cleanly
        column_names = [description[0] for description in cursor.description]
        formatted_results = [dict(zip(column_names, row)) for row in results]

        conn.close()
        return json.dumps(formatted_results, indent=2)

    except Exception as e:
        return f"Database error: {str(e)}"

Krótki blok rozpoczynający zamyka plik:

if __name__ == "__main__":
    setup_dummy_db()
    mcp.run()

Łączenie z Claude Desktop

Należy poinformować klienta, gdzie znajduje się serwer. W systemie Windows wejdź do folderu %APPDATA%\Claude, otwórz plik claude_desktop_config.json w edytorze i dodaj pozycję mcpServers.

Specyfika systemu Windows, która sprawia problemy: jeśli uruchamiasz program w środowisku wirtualnym takim jak Conda lub venv, podaj pełną ścieżkę do pliku python.exe tego środowiska zamiast prostego globalnego polecenia python, a także zapisuj wszystkie ścieżki z podwójnymi ukośnikami odwrotnymi.

{
  "mcpServers": {
    "enterprise-sql": {
      "command": "C:\\Users\\YourName\\.conda\\envs\\your_env\\python.exe",
      "args": [
        "D:\\Your\\Project\\Path\\mcp_server.py"
      ]
    }
  }
}

Zapisz plik i całkowicie ponownie uruchom Claude Desktop.

Widok działania

Otwórz Claude Desktop i poszukaj ikony młota (Narzędzia) w prawym dolnym rogu okna czatu. Narzędzie query_employee_database powinno już tam być zarejestrowane. Dobrym pierwszym testem jest zapytanie o to, kto zarabia najwięcej w tabeli pracowników i jaką ma funkcję.

Model czyta opis narzędzia, pisze sam kod SQL (SELECT * FROM employees ORDER BY salary DESC LIMIT 1;), przekazuje go do twojego serwera i przekształca wynik w odpowiedź w języku potocznym. Nie napisano żadnej specjalnej warstwy API, a schemat został odkryty dynamicznie.

To samo wzorce mają zastosowanie w Slacku, Jira, GitHubie lub dowolnej wewnętrznej API firmy, i to właśnie zmienia aspekty ekonomiczne tworzenia systemów agentowych w przedsiębiorstwie: każdy system jest pakowany tylko raz, a każdy kompatybilny asystent może go używać.