MCP-Oberfläche
- Resource:
verbs://schema - Tool:
get_database_schema() - Tool:
query_sql(sql)
Das Datenbankschema war explizit bekannt. Zusätzlich war die Semantik
von rowid 1..6 fest beschrieben.
Model Context Protocol · SQLite · Claude Code · Apache
Vom lokalen SQLite-MCP-Server über CGI-Zwischenstufen bis zum vollständig entfernten MCP-Server hinter Apache und systemd. Ziel aller vier Ausbaustufen ist, eine relationale Datenbank in natürlicher Sprache und ohne SQL-Kenntnisse abzufragen, daraus beliebige Sichten zusammenzustellen und die Ergebnisse durch ein Sprachmodell verständlich aufbereiten zu lassen.
Der hier in vier Ausbaustufen vorgestellte MCP-Server erlaubt es, eine SQLite-Datenbank über einen MCP-Client wie Claude Code in natürlicher Sprache abzufragen. Der Benutzer muss dazu keine SQL-Kenntnisse besitzen: Das LLM interpretiert die Aufgabe, ermittelt die benötigten Tabellen und Spalten, erzeugt die erforderlichen SQL-Statements und bereitet das Ergebnis als verständliche Sicht auf die Daten auf.
Als durchgängiges Beispiel dient eine Datenbank zur Konjugation altgriechischer
Verbklassen. Sie besteht aus acht vielspaltigen Tabellen mit identischer
Grundstruktur, jeweils einer Tabelle pro Verb- bzw. Konjugationsklasse
(paideuw, poiw, timw,
doulw, diwkw, grafw,
peithw und angellw). Die folgende Abbildung zeigt
als Ausschnitt die ersten Spalten der Tabelle für
παιδεύω (paideuw), so wie sie auf einer
klassischen Website dargestellt werden kann.
paideuw: Präsensformen der
Konjugationsklasse παιδεύω in einer klassischen tabellarischen Webdarstellung.




Die vier Varianten zeigen damit eine schrittweise Trennung von Client, MCP-Protokoll, Datenzugriff und Web-Infrastruktur. Gleichzeitig wird der MCP-Server immer generischer: vom explizit beschriebenen Schema bis zur dynamischen Schema-Ermittlung auf dem entfernten Server.
Claude Code startet einen lokalen MCP-Server über stdio; die SQLite-Datenbank liegt lokal.
Der MCP-Server bleibt lokal; die entfernte Datenbank wird nur über vorgegebene Tabellen-/Spaltenlisten abgefragt.
Claude entdeckt das Schema selbst und sendet beliebiges lesendes SQL an verbssql.sh.
MCP-Server und SQLite laufen gemeinsam auf dem Webserver; lokal bleibt nur die Client-Konfiguration.
Der einfachste vollständige MCP-Aufbau: Client, MCP-Server und Datenbank liegen lokal.
verbs://schemaget_database_schema()query_sql(sql)
Das Datenbankschema war explizit bekannt. Zusätzlich war die Semantik
von rowid 1..6 fest beschrieben.
file:...?mode=roPRAGMA query_only = ONSELECT/WITHSELECT
p."Präsens Indikativ",
t."Präsens Indikativ"
FROM poiw AS p
JOIN timw AS t ON t.rowid = p.rowid;
/usr/local/bin/verbs-mcp-server
/usr/local/lib/verbs-mcp/
server.py
.venv/
/usr/local/share/verbs-mcp/
verbsdynamic.db
Projektverzeichnis:
.mcp.json
#!/usr/bin/env python3
"""Minimal read-only MCP server for verbsdynamic.db."""
from __future__ import annotations
import json
import os
import re
import sqlite3
from pathlib import Path
from typing import Any
from mcp.server import MCPServer
SERVER_NAME = "verbsdynamic"
DEFAULT_DB = Path("/usr/local/share/verbs-mcp/verbsdynamic.db")
DB_PATH = Path(os.environ.get("VERBSDYNAMIC_DB", str(DEFAULT_DB)))
MAX_ROWS = 200
mcp = MCPServer(SERVER_NAME)
ROW_SEMANTICS = {
1: "1. Person Singular",
2: "2. Person Singular",
3: "3. Person Singular",
4: "1. Person Plural",
5: "2. Person Plural",
6: "3. Person Plural",
}
def _connect() -> sqlite3.Connection:
if not DB_PATH.is_file():
raise RuntimeError(f"SQLite database not found: {DB_PATH}")
# URI mode=ro makes the database read-only at SQLite level.
con = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True)
con.execute("PRAGMA query_only = ON")
con.set_authorizer(_authorizer)
return con
def _authorizer(action: int, arg1: str | None, arg2: str | None,
dbname: str | None, source: str | None) -> int:
"""Allow only operations needed for SELECT queries."""
allowed = {
sqlite3.SQLITE_SELECT,
sqlite3.SQLITE_READ,
sqlite3.SQLITE_FUNCTION,
}
if hasattr(sqlite3, "SQLITE_RECURSIVE"):
allowed.add(sqlite3.SQLITE_RECURSIVE)
return sqlite3.SQLITE_OK if action in allowed else sqlite3.SQLITE_DENY
def _first_sql_keyword(sql: str) -> str:
"""Return first SQL keyword after leading whitespace/comments."""
s = sql.lstrip()
while True:
if s.startswith("--"):
nl = s.find("\n")
s = "" if nl < 0 else s[nl + 1:].lstrip()
continue
if s.startswith("/*"):
end = s.find("*/", 2)
if end < 0:
return ""
s = s[end + 2:].lstrip()
continue
break
m = re.match(r"([A-Za-z]+)", s)
return m.group(1).upper() if m else ""
def _schema_dict() -> dict[str, Any]:
with _connect() as con:
tables = [
row[0]
for row in con.execute(
"SELECT name FROM sqlite_master "
"WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name"
)
]
result: dict[str, Any] = {
"database": str(DB_PATH),
"semantics": {
"one_table_per_verb": True,
"rowid": ROW_SEMANTICS,
"important": (
"All verb tables have the same column layout. "
"There is no explicit id/person column. SQLite rowid 1..6 "
"means 1sg, 2sg, 3sg, 1pl, 2pl, 3pl respectively. "
"Join different verb tables on rowid when comparing persons."
),
},
"tables": {},
}
for table in tables:
quoted = table.replace('"', '""')
cols = [
{"name": row[1], "type": row[2] or ""}
for row in con.execute(f'PRAGMA table_info("{quoted}")')
]
count = con.execute(f'SELECT count(*) FROM "{quoted}"').fetchone()[0]
result["tables"][table] = {"rows": count, "columns": cols}
return result
@mcp.resource("verbs://schema", mime_type="application/json")
def database_schema_resource() -> str:
"""Schema and row semantics of the Ancient Greek verb database."""
return json.dumps(_schema_dict(), ensure_ascii=False, indent=2)
@mcp.tool()
def get_database_schema() -> dict[str, Any]:
"""Return all verb tables, columns and row semantics before writing SQL.
Use this when the user's question requires database access and the exact
table/column names are not already known. Each table represents one verb.
rowid 1..6 corresponds to 1sg, 2sg, 3sg, 1pl, 2pl, 3pl.
"""
return _schema_dict()
@mcp.tool()
def query_sql(sql: str) -> dict[str, Any]:
"""Execute one read-only SELECT/WITH query against the verb database.
The database contains one table per verb and equal schemas across tables.
Use quoted identifiers for German column names, e.g.
SELECT rowid, "Präsens Indikativ" FROM poiw ORDER BY rowid.
To compare verbs, join tables on rowid. Only SELECT or WITH is accepted.
At most 200 result rows are returned.
"""
keyword = _first_sql_keyword(sql)
if keyword not in {"SELECT", "WITH"}:
raise ValueError("Only SELECT or WITH queries are allowed")
with _connect() as con:
cur = con.execute(sql)
columns = [d[0] for d in cur.description] if cur.description else []
rows = cur.fetchmany(MAX_ROWS + 1)
truncated = len(rows) > MAX_ROWS
if truncated:
rows = rows[:MAX_ROWS]
return {
"columns": columns,
"rows": [list(row) for row in rows],
"row_count": len(rows),
"truncated": truncated,
"max_rows": MAX_ROWS,
}
if __name__ == "__main__":
mcp.run(transport="stdio")
verbsdynamic.shDer MCP-Prozess bleibt lokal, aber Datenbank und SQL-Logik liegen nun auf dem Webserver.
query_verbs(
tables=["poiw", "timw"],
tempora=[
"Präsens Indikativ",
"Aorist Indikativ"
]
)
POST /cgi-bin/verbsdynamic.sh
tabelle=poiw,timw
tempus=Präsens Indikativ,Aorist Indikativ
/usr/local/bin/verbs-http-mcp-server
/usr/local/lib/verbs-http-mcp/
server.py
.venv/
.mcp.json
/cgi-bin/verbsdynamic.sh
/var/www/html/Public/Greek/verbsdynamic.db
Apache CGI-Konfiguration
#!/usr/bin/env python3
"""Minimal MCP bridge to the remote verbsdynamic CGI service."""
from __future__ import annotations
import html
import json
import os
import ssl
from html.parser import HTMLParser
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.parse import urlencode
from urllib.request import Request, urlopen
from mcp.server import MCPServer
SERVER_NAME = "verbsdynamic-remote"
DEFAULT_URL = "https://www.drbreinlinger.de/cgi-bin/verbsdynamic.sh"
SERVICE_URL = os.environ.get("VERBSDYNAMIC_URL", DEFAULT_URL)
TIMEOUT = float(os.environ.get("VERBSDYNAMIC_TIMEOUT", "15"))
mcp = MCPServer(SERVER_NAME)
VERBS = {
"paideuw": "παιδεὐω",
"poiw": "ποιῶ",
"timw": "τιμῶ",
"doulw": "δουλῶ",
"diwkw": "διώκω",
"grafw": "γρἀφω",
"peithw": "πείθω",
"angellw": "ἀγγέλλω",
}
TEMPORA = [
"Präsens Infinitiv",
"Präsens Partizip",
"Präsens Imperativ",
"Präsens Indikativ",
"Präsens Optativ",
"Präsens Konjunktiv",
"Futur Infinitiv",
"Futur Partizip",
"Futur Indikativ",
"Futur Optativ",
"Imperfekt Indikativ",
"Aorist Infinitiv",
"Aorist Partizip",
"Aorist Imperativ",
"Aorist Indikativ",
"Aorist Optativ",
"Aorist Konjunktiv",
"Perfekt Infinitiv",
"Perfekt Partizip",
"Perfekt Imperativ",
"Perfekt Indikativ",
"Perfekt Optativ",
"Perfekt Konjunktiv",
"Plusquamperfekt Indikativ",
"Futur II Infinitiv",
"Futur II Partizip",
"Futur II Indikativ",
"Futur II Optativ",
]
ROW_SEMANTICS = {
1: "1. Person Singular",
2: "2. Person Singular",
3: "3. Person Singular",
4: "1. Person Plural",
5: "2. Person Plural",
6: "3. Person Plural",
}
class _TableParser(HTMLParser):
def __init__(self) -> None:
super().__init__(convert_charrefs=True)
self.in_table = False
self.in_cell = False
self.cell_kind: str | None = None
self.cell_parts: list[str] = []
self.current_row: list[tuple[str, str]] = []
self.rows: list[list[tuple[str, str]]] = []
def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
tag = tag.lower()
if tag == "table" and not self.in_table:
self.in_table = True
elif self.in_table and tag == "tr":
self.current_row = []
elif self.in_table and tag in {"th", "td"}:
self.in_cell = True
self.cell_kind = tag
self.cell_parts = []
def handle_data(self, data: str) -> None:
if self.in_cell:
self.cell_parts.append(data)
def handle_endtag(self, tag: str) -> None:
tag = tag.lower()
if self.in_table and tag in {"th", "td"} and self.in_cell:
text = html.unescape("".join(self.cell_parts)).strip()
self.current_row.append((self.cell_kind or "td", text))
self.in_cell = False
self.cell_kind = None
self.cell_parts = []
elif self.in_table and tag == "tr":
if self.current_row:
self.rows.append(self.current_row)
self.current_row = []
elif self.in_table and tag == "table":
self.in_table = False
def _options() -> dict[str, Any]:
return {
"service_url": SERVICE_URL,
"verbs": VERBS,
"tempora": TEMPORA,
"row_semantics": ROW_SEMANTICS,
"protocol": {
"method": "POST",
"content_type": "application/x-www-form-urlencoded; charset=UTF-8",
"fields": {
"tabelle": "comma-separated table names",
"tempus": "comma-separated column names",
},
},
}
def _validate(tables: list[str], tempora: list[str]) -> tuple[list[str], list[str]]:
if not tables:
raise ValueError("At least one table/verb is required")
if not tempora:
raise ValueError("At least one tempus/column is required")
bad_tables = [x for x in tables if x not in VERBS]
if bad_tables:
raise ValueError(f"Unknown table(s): {', '.join(bad_tables)}")
bad_tempora = [x for x in tempora if x not in TEMPORA]
if bad_tempora:
raise ValueError(f"Unknown tempus/column(s): {', '.join(bad_tempora)}")
# Preserve requested order while suppressing duplicates.
tables = list(dict.fromkeys(tables))
tempora = list(dict.fromkeys(tempora))
return tables, tempora
def _ssl_context() -> ssl.SSLContext:
"""Create a verifying TLS context suitable for Zscaler-intercepted HTTPS.
VERIFY_X509_PARTIAL_CHAIN keeps certificate and hostname verification
enabled, but allows chain building to terminate at a trusted intermediate
CA in the local trust store.
"""
context = ssl.create_default_context()
if hasattr(ssl, "VERIFY_X509_PARTIAL_CHAIN"):
context.verify_flags |= ssl.VERIFY_X509_PARTIAL_CHAIN
return context
def _post_query(tables: list[str], tempora: list[str]) -> str:
data = urlencode({
"tabelle": ",".join(tables),
"tempus": ",".join(tempora),
}).encode("utf-8")
request = Request(
SERVICE_URL,
data=data,
method="POST",
headers={
"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
"User-Agent": "verbsdynamic-mcp/1.0",
},
)
try:
with urlopen(request, timeout=TIMEOUT, context=_ssl_context()) as response:
charset = response.headers.get_content_charset() or "utf-8"
return response.read().decode(charset, errors="replace")
except HTTPError as exc:
body = exc.read().decode("utf-8", errors="replace")
raise RuntimeError(f"CGI returned HTTP {exc.code}: {body[:500]}") from exc
except URLError as exc:
raise RuntimeError(f"Cannot reach CGI service {SERVICE_URL}: {exc.reason}") from exc
def _parse_table(document: str) -> dict[str, Any]:
parser = _TableParser()
parser.feed(document)
if not parser.rows:
raise RuntimeError(f"CGI response contains no HTML table: {document[:500]}")
first = parser.rows[0]
if first and all(kind == "th" for kind, _ in first):
columns = [text for _, text in first]
data_rows = parser.rows[1:]
else:
columns = []
data_rows = parser.rows
rows = [[text for _, text in row] for row in data_rows]
return {
"columns": columns,
"rows": rows,
"row_count": len(rows),
}
@mcp.resource("verbsremote://schema", mime_type="application/json")
def query_options_resource() -> str:
"""Available verb tables, column names and CGI protocol."""
return json.dumps(_options(), ensure_ascii=False, indent=2)
@mcp.tool()
def get_query_options() -> dict[str, Any]:
"""Return valid verb/table names and tempus/column names for the remote service."""
return _options()
@mcp.tool()
def query_verbs(tables: list[str], tempora: list[str]) -> dict[str, Any]:
"""Query Ancient Greek verb forms through the remote verbsdynamic CGI service.
'tables' contains one or more table identifiers such as ["poiw", "timw"].
'tempora' contains one or more exact column names such as
["Präsens Indikativ", "Aorist Indikativ"].
The tool sends the request as application/x-www-form-urlencoded using the
CGI fields 'tabelle' and 'tempus', each as a comma-separated list.
It converts the returned HTML table into structured columns and rows.
"""
tables, tempora = _validate(tables, tempora)
document = _post_query(tables, tempora)
result = _parse_table(document)
result.update({
"tables": tables,
"tempora": tempora,
"row_semantics": ROW_SEMANTICS,
"service_url": SERVICE_URL,
})
return result
if __name__ == "__main__":
mcp.run(transport="stdio")
verbssql.shDer MCP-Server kennt keine Verbklassen oder Spalten mehr. Claude erzeugt direkt lesendes SQL.
SELECT name
FROM sqlite_master
WHERE type='table'
ORDER BY name;
SELECT *
FROM pragma_table_info('poiw');
#!/usr/bin/env python3
"""MCP bridge for arbitrary read-only SQL queries via verbssql.sh."""
from __future__ import annotations
import html
import json
import os
import re
import ssl
from html.parser import HTMLParser
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.parse import urlencode
from urllib.request import Request, urlopen
from mcp.server import MCPServer
SERVER_NAME = "verbssql-remote"
DEFAULT_URL = "https://www.drbreinlinger.de/cgi-bin/verbssql.sh"
SERVICE_URL = os.environ.get("VERBSSQL_URL", DEFAULT_URL)
TIMEOUT = float(os.environ.get("VERBSSQL_TIMEOUT", "15"))
mcp = MCPServer(SERVER_NAME)
READ_ONLY_START = re.compile(r"^\s*(?:SELECT|WITH)\b", re.IGNORECASE)
class _TableParser(HTMLParser):
def __init__(self) -> None:
super().__init__(convert_charrefs=True)
self.in_table = False
self.in_cell = False
self.cell_kind: str | None = None
self.cell_parts: list[str] = []
self.current_row: list[tuple[str, str]] = []
self.rows: list[list[tuple[str, str]]] = []
def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
tag = tag.lower()
if tag == "table" and not self.in_table:
self.in_table = True
elif self.in_table and tag == "tr":
self.current_row = []
elif self.in_table and tag in {"th", "td"}:
self.in_cell = True
self.cell_kind = tag
self.cell_parts = []
def handle_data(self, data: str) -> None:
if self.in_cell:
self.cell_parts.append(data)
def handle_endtag(self, tag: str) -> None:
tag = tag.lower()
if self.in_table and tag in {"th", "td"} and self.in_cell:
text = html.unescape("".join(self.cell_parts)).strip()
self.current_row.append((self.cell_kind or "td", text))
self.in_cell = False
self.cell_kind = None
self.cell_parts = []
elif self.in_table and tag == "tr":
if self.current_row:
self.rows.append(self.current_row)
self.current_row = []
elif self.in_table and tag == "table":
self.in_table = False
def _ssl_context() -> ssl.SSLContext:
context = ssl.create_default_context()
if hasattr(ssl, "VERIFY_X509_PARTIAL_CHAIN"):
context.verify_flags |= ssl.VERIFY_X509_PARTIAL_CHAIN
return context
def _parse_table(document: str) -> dict[str, Any]:
parser = _TableParser()
parser.feed(document)
if not parser.rows:
# Preserve useful CGI error text if the response is not a table.
text = re.sub(r"<[^>]+>", " ", document)
text = html.unescape(re.sub(r"\s+", " ", text)).strip()
raise RuntimeError(f"CGI response contains no HTML table: {text[:500]}")
first = parser.rows[0]
if first and all(kind == "th" for kind, _ in first):
columns = [text for _, text in first]
data_rows = parser.rows[1:]
else:
columns = []
data_rows = parser.rows
rows = [[text for _, text in row] for row in data_rows]
return {"columns": columns, "rows": rows, "row_count": len(rows)}
def _post_sql(sql: str) -> str:
data = urlencode({"sql": sql}).encode("utf-8")
request = Request(
SERVICE_URL,
data=data,
method="POST",
headers={
"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
"User-Agent": "verbssql-mcp/1.0",
},
)
try:
with urlopen(request, timeout=TIMEOUT, context=_ssl_context()) as response:
charset = response.headers.get_content_charset() or "utf-8"
return response.read().decode(charset, errors="replace")
except HTTPError as exc:
body = exc.read().decode("utf-8", errors="replace")
raise RuntimeError(f"CGI returned HTTP {exc.code}: {body[:500]}") from exc
except URLError as exc:
raise RuntimeError(f"Cannot reach CGI service {SERVICE_URL}: {exc.reason}") from exc
@mcp.resource("verbssql://info", mime_type="application/json")
def service_info_resource() -> str:
return json.dumps(
{
"service_url": SERVICE_URL,
"method": "POST",
"content_type": "application/x-www-form-urlencoded; charset=UTF-8",
"field": "sql",
"purpose": "Execute arbitrary read-only SQL against the remote Greek verbs SQLite database.",
},
ensure_ascii=False,
indent=2,
)
@mcp.tool()
def query_sql(sql: str) -> dict[str, Any]:
"""Execute a read-only SQL query through the remote verbssql CGI service.
Send a SELECT or WITH query exactly as SQL. The CGI service performs the
authoritative read-only validation. This MCP bridge additionally rejects
statements that do not begin with SELECT or WITH, then POSTs the SQL in the
form field 'sql' and converts the returned HTML table to structured data.
The MCP server does not contain a hard-coded database schema. If table or
column names are unknown, discover the current SQLite schema through this
same tool before composing the actual query, for example:
SELECT name
FROM sqlite_master
WHERE type='table'
ORDER BY name;
and for a specific table:
SELECT *
FROM pragma_table_info('poiw');
Prefer schema discovery over guessing table or column names.
"""
if not isinstance(sql, str) or not sql.strip():
raise ValueError("SQL query must not be empty")
if not READ_ONLY_START.match(sql):
raise ValueError("Only read-only SQL beginning with SELECT or WITH is allowed")
document = _post_sql(sql)
result = _parse_table(document)
result.update({"sql": sql, "service_url": SERVICE_URL})
return result
if __name__ == "__main__":
mcp.run(transport="stdio")
Die finale Architektur: MCP-Server und SQLite liegen gemeinsam auf www.drbreinlinger.de.
ProxyPreserveHost On
ProxyPass "/mcp/verbs" \
"http://127.0.0.1:8765/mcp"
ProxyPassReverse "/mcp/verbs" \
"http://127.0.0.1:8765/mcp"
User=mcp
Group=mcp
SupplementaryGroups=www-data
ExecStart=.../.venv/bin/python \
.../server.py
Restart=on-failure
.mcp.jsonAuf dem Client ist kein lokaler MCP-Server mehr erforderlich. Die folgende Projektkonfiguration genügt; das Bearer-Token wird über VERBS_MCP_TOKEN bereitgestellt.
{
"mcpServers": {
"greek-verbs": {
"type": "http",
"url": "https://www.drbreinlinger.de/mcp/verbs",
"headers": {
"Authorization": "Bearer ${VERBS_MCP_TOKEN}"
}
}
}
}
Projekt/.mcp.json
Umgebung:
VERBS_MCP_TOKEN=...
Kein lokales venv
Keine lokale DB
Kein lokaler MCP-Prozess
/usr/local/lib/verbs-remote-mcp/
server.py
.venv/
/etc/systemd/system/
verbs-remote-mcp.service
Apache VirtualHost:
/mcp/verbs → 127.0.0.1:8765/mcp
/var/www/html/Public/Greek/
verbsdynamic.db
.mcp.json und das Zugriffstoken.#!/usr/bin/env python3
"""Remote MCP server for read-only access to verbsdynamic.db."""
from __future__ import annotations
import os
import re
import sqlite3
import time
from pathlib import Path
from typing import Any
from mcp.server import MCPServer
from mcp.server.transport_security import TransportSecuritySettings
SERVER_NAME = "greek-verbs-sql"
DB_PATH = Path(os.environ.get(
"VERBS_MCP_DB",
"/var/www/html/Public/Greek/verbsdynamic.db",
))
HOST = os.environ.get("VERBS_MCP_HOST", "127.0.0.1")
PORT = int(os.environ.get("VERBS_MCP_PORT", "8765"))
PUBLIC_HOST = os.environ.get("VERBS_MCP_PUBLIC_HOST", "www.drbreinlinger.de")
MAX_ROWS = int(os.environ.get("VERBS_MCP_MAX_ROWS", "500"))
MAX_SECONDS = float(os.environ.get("VERBS_MCP_MAX_SECONDS", "3.0"))
READ_ONLY_START = re.compile(r"^\s*(?:SELECT|WITH)\b", re.IGNORECASE)
mcp = MCPServer(
SERVER_NAME,
instructions=(
"This server exposes a read-only SQLite database through query_sql(). "
"The schema is deliberately not hard-coded. If table or column names "
"are unknown, discover them with SELECTs against sqlite_master and "
"table-valued pragma functions such as pragma_table_info(). "
"Prefer schema discovery over guessing identifiers."
),
)
# Read-only schema PRAGMAs that SQLite may invoke internally for table-valued
# pragma functions such as SELECT * FROM pragma_table_info('poiw').
SAFE_PRAGMAS = {
"table_info",
"table_xinfo",
"index_list",
"index_info",
"index_xinfo",
"foreign_key_list",
}
def _authorizer(
action: int,
arg1: str | None,
arg2: str | None,
dbname: str | None,
source: str | None,
) -> int:
allowed = {
sqlite3.SQLITE_SELECT,
sqlite3.SQLITE_READ,
sqlite3.SQLITE_FUNCTION,
}
if hasattr(sqlite3, "SQLITE_RECURSIVE"):
allowed.add(sqlite3.SQLITE_RECURSIVE)
if action in allowed:
return sqlite3.SQLITE_OK
if action == sqlite3.SQLITE_PRAGMA and (arg1 or "").lower() in SAFE_PRAGMAS:
return sqlite3.SQLITE_OK
return sqlite3.SQLITE_DENY
def _connect() -> sqlite3.Connection:
if not DB_PATH.is_file():
raise RuntimeError(f"SQLite database not found: {DB_PATH}")
con = sqlite3.connect(
f"file:{DB_PATH}?mode=ro",
uri=True,
timeout=2.0,
)
con.execute("PRAGMA query_only = ON")
con.set_authorizer(_authorizer)
deadline = time.monotonic() + MAX_SECONDS
def progress() -> int:
return 1 if time.monotonic() > deadline else 0
# Called every 1000 SQLite virtual-machine instructions.
con.set_progress_handler(progress, 1000)
return con
@mcp.tool()
def query_sql(sql: str) -> dict[str, Any]:
"""Execute one read-only SQLite SELECT/WITH query.
The database schema is not hard-coded into this MCP server. When tables or
columns are unknown, discover the current schema through this same tool.
Examples:
SELECT name
FROM sqlite_master
WHERE type='table'
ORDER BY name;
SELECT *
FROM pragma_table_info('poiw');
Then compose the actual query from the discovered identifiers. Prefer
schema discovery over guessing table or column names.
Only SELECT or WITH statements are accepted. SQLite itself is additionally
opened read-only, query_only is enabled, and an authorizer rejects writes
and unsafe PRAGMAs. Results are capped at MAX_ROWS and execution time is
bounded.
"""
if not isinstance(sql, str) or not sql.strip():
raise ValueError("SQL query must not be empty")
if not READ_ONLY_START.match(sql):
raise ValueError("Only read-only SQL beginning with SELECT or WITH is allowed")
try:
with _connect() as con:
cur = con.execute(sql)
columns = [d[0] for d in cur.description] if cur.description else []
rows = cur.fetchmany(MAX_ROWS + 1)
except sqlite3.OperationalError as exc:
if "interrupted" in str(exc).lower():
raise RuntimeError(
f"SQLite query exceeded the {MAX_SECONDS:g}s execution limit"
) from exc
raise
truncated = len(rows) > MAX_ROWS
if truncated:
rows = rows[:MAX_ROWS]
return {
"columns": columns,
"rows": [list(row) for row in rows],
"row_count": len(rows),
"truncated": truncated,
"max_rows": MAX_ROWS,
}
if __name__ == "__main__":
security = TransportSecuritySettings(
allowed_hosts=[
PUBLIC_HOST,
f"{PUBLIC_HOST}:*",
"127.0.0.1",
"127.0.0.1:*",
"localhost",
"localhost:*",
],
allowed_origins=[
f"https://{PUBLIC_HOST}",
],
)
mcp.run(
transport="streamable-http",
host=HOST,
port=PORT,
streamable_http_path="/mcp",
transport_security=security,
)
unzip verbs-remote-mcp-mcpuser.zip
cd verbs-remote-mcp-mcpuser
sudo ./install.sh
mcp prüfen
sudo -u mcp test -r \
/var/www/html/Public/Greek/verbsdynamic.db \
&& echo readable
sudo systemctl enable --now verbs-remote-mcp
systemctl status verbs-remote-mcp
sudo a2enmod proxy proxy_http
sudo apachectl configtest
sudo systemctl reload apache2
export VERBS_MCP_TOKEN='...'
claude mcp list
| Variante | MCP-Ort | DB-Ort | Transport | Schema | Zwischenebene |
|---|---|---|---|---|---|
| A · Local MCP / Local DB (generic) | Client | Client | stdio | fest + Resource | — |
| B · Local MCP / Remote DB (limited) | Client | Server | stdio + HTTPS | fachlich fest | CGI + HTML |
| C · Local MCP / Remote DB (generic) | Client | Server | stdio + HTTPS | dynamisch | CGI + HTML |
| D · Remote MCP / Remote DB (generic) | Server | Server | HTTPS / MCP | dynamisch | Apache reverse proxy |
Model Context Protocol · SQLite · Claude Code · Apache
From a local SQLite MCP server through intermediate CGI stages to a fully remote MCP server behind Apache and systemd. The goal of all four stages is to query a relational database in natural language without requiring SQL knowledge, build arbitrary views of the data, and have a language model present the results in a useful form.
The MCP server presented here in four evolution stages makes it possible to query an SQLite database in natural language through an MCP client such as Claude Code. The user does not need SQL knowledge: the LLM interprets the task, determines the required tables and columns, generates the necessary SQL statements, and presents the result as a useful view of the data.
The running example is a database for the conjugation of Ancient Greek verb
classes. It consists of eight wide tables with the same basic structure, one
table for each verb or conjugation class
(paideuw, poiw, timw,
doulw, diwkw, grafw,
peithw, and angellw). The following image shows
the first columns of the table for παιδεύω
(paideuw) as they might be displayed on a conventional website.
paideuw: present-tense forms of the
παιδεύω conjugation class in a conventional tabular web view.




The four variants therefore show a gradual separation of client, MCP protocol, data access, and web infrastructure. At the same time, the MCP server becomes increasingly generic: from an explicitly described schema to dynamic schema discovery on the remote server.
Claude Code starts a local MCP server via stdio; the SQLite database is local.
The MCP server remains local; the remote database is queried only through predefined table and column lists.
Claude discovers the schema and sends arbitrary read-only SQL to verbssql.sh.
MCP server and SQLite run together on the web server; only the client configuration remains local.
The simplest complete MCP setup: client, MCP server, and database are all local.
verbs://schemaget_database_schema()query_sql(sql)
The database schema was explicitly known. In addition, the semantics of
rowid 1..6 were explicitly defined.
file:...?mode=roPRAGMA query_only = ONSELECT/WITHSELECT
p."Präsens Indikativ",
t."Präsens Indikativ"
FROM poiw AS p
JOIN timw AS t ON t.rowid = p.rowid;
/usr/local/bin/verbs-mcp-server
/usr/local/lib/verbs-mcp/
server.py
.venv/
/usr/local/share/verbs-mcp/
verbsdynamic.db
Project directory:
.mcp.json
#!/usr/bin/env python3
"""Minimal read-only MCP server for verbsdynamic.db."""
from __future__ import annotations
import json
import os
import re
import sqlite3
from pathlib import Path
from typing import Any
from mcp.server import MCPServer
SERVER_NAME = "verbsdynamic"
DEFAULT_DB = Path("/usr/local/share/verbs-mcp/verbsdynamic.db")
DB_PATH = Path(os.environ.get("VERBSDYNAMIC_DB", str(DEFAULT_DB)))
MAX_ROWS = 200
mcp = MCPServer(SERVER_NAME)
ROW_SEMANTICS = {
1: "1. Person Singular",
2: "2. Person Singular",
3: "3. Person Singular",
4: "1. Person Plural",
5: "2. Person Plural",
6: "3. Person Plural",
}
def _connect() -> sqlite3.Connection:
if not DB_PATH.is_file():
raise RuntimeError(f"SQLite database not found: {DB_PATH}")
# URI mode=ro makes the database read-only at SQLite level.
con = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True)
con.execute("PRAGMA query_only = ON")
con.set_authorizer(_authorizer)
return con
def _authorizer(action: int, arg1: str | None, arg2: str | None,
dbname: str | None, source: str | None) -> int:
"""Allow only operations needed for SELECT queries."""
allowed = {
sqlite3.SQLITE_SELECT,
sqlite3.SQLITE_READ,
sqlite3.SQLITE_FUNCTION,
}
if hasattr(sqlite3, "SQLITE_RECURSIVE"):
allowed.add(sqlite3.SQLITE_RECURSIVE)
return sqlite3.SQLITE_OK if action in allowed else sqlite3.SQLITE_DENY
def _first_sql_keyword(sql: str) -> str:
"""Return first SQL keyword after leading whitespace/comments."""
s = sql.lstrip()
while True:
if s.startswith("--"):
nl = s.find("\n")
s = "" if nl < 0 else s[nl + 1:].lstrip()
continue
if s.startswith("/*"):
end = s.find("*/", 2)
if end < 0:
return ""
s = s[end + 2:].lstrip()
continue
break
m = re.match(r"([A-Za-z]+)", s)
return m.group(1).upper() if m else ""
def _schema_dict() -> dict[str, Any]:
with _connect() as con:
tables = [
row[0]
for row in con.execute(
"SELECT name FROM sqlite_master "
"WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name"
)
]
result: dict[str, Any] = {
"database": str(DB_PATH),
"semantics": {
"one_table_per_verb": True,
"rowid": ROW_SEMANTICS,
"important": (
"All verb tables have the same column layout. "
"There is no explicit id/person column. SQLite rowid 1..6 "
"means 1sg, 2sg, 3sg, 1pl, 2pl, 3pl respectively. "
"Join different verb tables on rowid when comparing persons."
),
},
"tables": {},
}
for table in tables:
quoted = table.replace('"', '""')
cols = [
{"name": row[1], "type": row[2] or ""}
for row in con.execute(f'PRAGMA table_info("{quoted}")')
]
count = con.execute(f'SELECT count(*) FROM "{quoted}"').fetchone()[0]
result["tables"][table] = {"rows": count, "columns": cols}
return result
@mcp.resource("verbs://schema", mime_type="application/json")
def database_schema_resource() -> str:
"""Schema and row semantics of the Ancient Greek verb database."""
return json.dumps(_schema_dict(), ensure_ascii=False, indent=2)
@mcp.tool()
def get_database_schema() -> dict[str, Any]:
"""Return all verb tables, columns and row semantics before writing SQL.
Use this when the user's question requires database access and the exact
table/column names are not already known. Each table represents one verb.
rowid 1..6 corresponds to 1sg, 2sg, 3sg, 1pl, 2pl, 3pl.
"""
return _schema_dict()
@mcp.tool()
def query_sql(sql: str) -> dict[str, Any]:
"""Execute one read-only SELECT/WITH query against the verb database.
The database contains one table per verb and equal schemas across tables.
Use quoted identifiers for German column names, e.g.
SELECT rowid, "Präsens Indikativ" FROM poiw ORDER BY rowid.
To compare verbs, join tables on rowid. Only SELECT or WITH is accepted.
At most 200 result rows are returned.
"""
keyword = _first_sql_keyword(sql)
if keyword not in {"SELECT", "WITH"}:
raise ValueError("Only SELECT or WITH queries are allowed")
with _connect() as con:
cur = con.execute(sql)
columns = [d[0] for d in cur.description] if cur.description else []
rows = cur.fetchmany(MAX_ROWS + 1)
truncated = len(rows) > MAX_ROWS
if truncated:
rows = rows[:MAX_ROWS]
return {
"columns": columns,
"rows": [list(row) for row in rows],
"row_count": len(rows),
"truncated": truncated,
"max_rows": MAX_ROWS,
}
if __name__ == "__main__":
mcp.run(transport="stdio")
verbsdynamic.shThe MCP process remains local, but the database and SQL logic now reside on the web server.
query_verbs(
tables=["poiw", "timw"],
tempora=[
"Präsens Indikativ",
"Aorist Indikativ"
]
)
POST /cgi-bin/verbsdynamic.sh
tabelle=poiw,timw
tempus=Präsens Indikativ,Aorist Indikativ
/usr/local/bin/verbs-http-mcp-server
/usr/local/lib/verbs-http-mcp/
server.py
.venv/
.mcp.json
/cgi-bin/verbsdynamic.sh
/var/www/html/Public/Greek/verbsdynamic.db
Apache CGI-Konfiguration
#!/usr/bin/env python3
"""Minimal MCP bridge to the remote verbsdynamic CGI service."""
from __future__ import annotations
import html
import json
import os
import ssl
from html.parser import HTMLParser
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.parse import urlencode
from urllib.request import Request, urlopen
from mcp.server import MCPServer
SERVER_NAME = "verbsdynamic-remote"
DEFAULT_URL = "https://www.drbreinlinger.de/cgi-bin/verbsdynamic.sh"
SERVICE_URL = os.environ.get("VERBSDYNAMIC_URL", DEFAULT_URL)
TIMEOUT = float(os.environ.get("VERBSDYNAMIC_TIMEOUT", "15"))
mcp = MCPServer(SERVER_NAME)
VERBS = {
"paideuw": "παιδεὐω",
"poiw": "ποιῶ",
"timw": "τιμῶ",
"doulw": "δουλῶ",
"diwkw": "διώκω",
"grafw": "γρἀφω",
"peithw": "πείθω",
"angellw": "ἀγγέλλω",
}
TEMPORA = [
"Präsens Infinitiv",
"Präsens Partizip",
"Präsens Imperativ",
"Präsens Indikativ",
"Präsens Optativ",
"Präsens Konjunktiv",
"Futur Infinitiv",
"Futur Partizip",
"Futur Indikativ",
"Futur Optativ",
"Imperfekt Indikativ",
"Aorist Infinitiv",
"Aorist Partizip",
"Aorist Imperativ",
"Aorist Indikativ",
"Aorist Optativ",
"Aorist Konjunktiv",
"Perfekt Infinitiv",
"Perfekt Partizip",
"Perfekt Imperativ",
"Perfekt Indikativ",
"Perfekt Optativ",
"Perfekt Konjunktiv",
"Plusquamperfekt Indikativ",
"Futur II Infinitiv",
"Futur II Partizip",
"Futur II Indikativ",
"Futur II Optativ",
]
ROW_SEMANTICS = {
1: "1. Person Singular",
2: "2. Person Singular",
3: "3. Person Singular",
4: "1. Person Plural",
5: "2. Person Plural",
6: "3. Person Plural",
}
class _TableParser(HTMLParser):
def __init__(self) -> None:
super().__init__(convert_charrefs=True)
self.in_table = False
self.in_cell = False
self.cell_kind: str | None = None
self.cell_parts: list[str] = []
self.current_row: list[tuple[str, str]] = []
self.rows: list[list[tuple[str, str]]] = []
def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
tag = tag.lower()
if tag == "table" and not self.in_table:
self.in_table = True
elif self.in_table and tag == "tr":
self.current_row = []
elif self.in_table and tag in {"th", "td"}:
self.in_cell = True
self.cell_kind = tag
self.cell_parts = []
def handle_data(self, data: str) -> None:
if self.in_cell:
self.cell_parts.append(data)
def handle_endtag(self, tag: str) -> None:
tag = tag.lower()
if self.in_table and tag in {"th", "td"} and self.in_cell:
text = html.unescape("".join(self.cell_parts)).strip()
self.current_row.append((self.cell_kind or "td", text))
self.in_cell = False
self.cell_kind = None
self.cell_parts = []
elif self.in_table and tag == "tr":
if self.current_row:
self.rows.append(self.current_row)
self.current_row = []
elif self.in_table and tag == "table":
self.in_table = False
def _options() -> dict[str, Any]:
return {
"service_url": SERVICE_URL,
"verbs": VERBS,
"tempora": TEMPORA,
"row_semantics": ROW_SEMANTICS,
"protocol": {
"method": "POST",
"content_type": "application/x-www-form-urlencoded; charset=UTF-8",
"fields": {
"tabelle": "comma-separated table names",
"tempus": "comma-separated column names",
},
},
}
def _validate(tables: list[str], tempora: list[str]) -> tuple[list[str], list[str]]:
if not tables:
raise ValueError("At least one table/verb is required")
if not tempora:
raise ValueError("At least one tempus/column is required")
bad_tables = [x for x in tables if x not in VERBS]
if bad_tables:
raise ValueError(f"Unknown table(s): {', '.join(bad_tables)}")
bad_tempora = [x for x in tempora if x not in TEMPORA]
if bad_tempora:
raise ValueError(f"Unknown tempus/column(s): {', '.join(bad_tempora)}")
# Preserve requested order while suppressing duplicates.
tables = list(dict.fromkeys(tables))
tempora = list(dict.fromkeys(tempora))
return tables, tempora
def _ssl_context() -> ssl.SSLContext:
"""Create a verifying TLS context suitable for Zscaler-intercepted HTTPS.
VERIFY_X509_PARTIAL_CHAIN keeps certificate and hostname verification
enabled, but allows chain building to terminate at a trusted intermediate
CA in the local trust store.
"""
context = ssl.create_default_context()
if hasattr(ssl, "VERIFY_X509_PARTIAL_CHAIN"):
context.verify_flags |= ssl.VERIFY_X509_PARTIAL_CHAIN
return context
def _post_query(tables: list[str], tempora: list[str]) -> str:
data = urlencode({
"tabelle": ",".join(tables),
"tempus": ",".join(tempora),
}).encode("utf-8")
request = Request(
SERVICE_URL,
data=data,
method="POST",
headers={
"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
"User-Agent": "verbsdynamic-mcp/1.0",
},
)
try:
with urlopen(request, timeout=TIMEOUT, context=_ssl_context()) as response:
charset = response.headers.get_content_charset() or "utf-8"
return response.read().decode(charset, errors="replace")
except HTTPError as exc:
body = exc.read().decode("utf-8", errors="replace")
raise RuntimeError(f"CGI returned HTTP {exc.code}: {body[:500]}") from exc
except URLError as exc:
raise RuntimeError(f"Cannot reach CGI service {SERVICE_URL}: {exc.reason}") from exc
def _parse_table(document: str) -> dict[str, Any]:
parser = _TableParser()
parser.feed(document)
if not parser.rows:
raise RuntimeError(f"CGI response contains no HTML table: {document[:500]}")
first = parser.rows[0]
if first and all(kind == "th" for kind, _ in first):
columns = [text for _, text in first]
data_rows = parser.rows[1:]
else:
columns = []
data_rows = parser.rows
rows = [[text for _, text in row] for row in data_rows]
return {
"columns": columns,
"rows": rows,
"row_count": len(rows),
}
@mcp.resource("verbsremote://schema", mime_type="application/json")
def query_options_resource() -> str:
"""Available verb tables, column names and CGI protocol."""
return json.dumps(_options(), ensure_ascii=False, indent=2)
@mcp.tool()
def get_query_options() -> dict[str, Any]:
"""Return valid verb/table names and tempus/column names for the remote service."""
return _options()
@mcp.tool()
def query_verbs(tables: list[str], tempora: list[str]) -> dict[str, Any]:
"""Query Ancient Greek verb forms through the remote verbsdynamic CGI service.
'tables' contains one or more table identifiers such as ["poiw", "timw"].
'tempora' contains one or more exact column names such as
["Präsens Indikativ", "Aorist Indikativ"].
The tool sends the request as application/x-www-form-urlencoded using the
CGI fields 'tabelle' and 'tempus', each as a comma-separated list.
It converts the returned HTML table into structured columns and rows.
"""
tables, tempora = _validate(tables, tempora)
document = _post_query(tables, tempora)
result = _parse_table(document)
result.update({
"tables": tables,
"tempora": tempora,
"row_semantics": ROW_SEMANTICS,
"service_url": SERVICE_URL,
})
return result
if __name__ == "__main__":
mcp.run(transport="stdio")
verbssql.shThe MCP server no longer knows any verb classes or columns. Claude generates read-only SQL directly.
SELECT name
FROM sqlite_master
WHERE type='table'
ORDER BY name;
SELECT *
FROM pragma_table_info('poiw');
#!/usr/bin/env python3
"""MCP bridge for arbitrary read-only SQL queries via verbssql.sh."""
from __future__ import annotations
import html
import json
import os
import re
import ssl
from html.parser import HTMLParser
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.parse import urlencode
from urllib.request import Request, urlopen
from mcp.server import MCPServer
SERVER_NAME = "verbssql-remote"
DEFAULT_URL = "https://www.drbreinlinger.de/cgi-bin/verbssql.sh"
SERVICE_URL = os.environ.get("VERBSSQL_URL", DEFAULT_URL)
TIMEOUT = float(os.environ.get("VERBSSQL_TIMEOUT", "15"))
mcp = MCPServer(SERVER_NAME)
READ_ONLY_START = re.compile(r"^\s*(?:SELECT|WITH)\b", re.IGNORECASE)
class _TableParser(HTMLParser):
def __init__(self) -> None:
super().__init__(convert_charrefs=True)
self.in_table = False
self.in_cell = False
self.cell_kind: str | None = None
self.cell_parts: list[str] = []
self.current_row: list[tuple[str, str]] = []
self.rows: list[list[tuple[str, str]]] = []
def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
tag = tag.lower()
if tag == "table" and not self.in_table:
self.in_table = True
elif self.in_table and tag == "tr":
self.current_row = []
elif self.in_table and tag in {"th", "td"}:
self.in_cell = True
self.cell_kind = tag
self.cell_parts = []
def handle_data(self, data: str) -> None:
if self.in_cell:
self.cell_parts.append(data)
def handle_endtag(self, tag: str) -> None:
tag = tag.lower()
if self.in_table and tag in {"th", "td"} and self.in_cell:
text = html.unescape("".join(self.cell_parts)).strip()
self.current_row.append((self.cell_kind or "td", text))
self.in_cell = False
self.cell_kind = None
self.cell_parts = []
elif self.in_table and tag == "tr":
if self.current_row:
self.rows.append(self.current_row)
self.current_row = []
elif self.in_table and tag == "table":
self.in_table = False
def _ssl_context() -> ssl.SSLContext:
context = ssl.create_default_context()
if hasattr(ssl, "VERIFY_X509_PARTIAL_CHAIN"):
context.verify_flags |= ssl.VERIFY_X509_PARTIAL_CHAIN
return context
def _parse_table(document: str) -> dict[str, Any]:
parser = _TableParser()
parser.feed(document)
if not parser.rows:
# Preserve useful CGI error text if the response is not a table.
text = re.sub(r"<[^>]+>", " ", document)
text = html.unescape(re.sub(r"\s+", " ", text)).strip()
raise RuntimeError(f"CGI response contains no HTML table: {text[:500]}")
first = parser.rows[0]
if first and all(kind == "th" for kind, _ in first):
columns = [text for _, text in first]
data_rows = parser.rows[1:]
else:
columns = []
data_rows = parser.rows
rows = [[text for _, text in row] for row in data_rows]
return {"columns": columns, "rows": rows, "row_count": len(rows)}
def _post_sql(sql: str) -> str:
data = urlencode({"sql": sql}).encode("utf-8")
request = Request(
SERVICE_URL,
data=data,
method="POST",
headers={
"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
"User-Agent": "verbssql-mcp/1.0",
},
)
try:
with urlopen(request, timeout=TIMEOUT, context=_ssl_context()) as response:
charset = response.headers.get_content_charset() or "utf-8"
return response.read().decode(charset, errors="replace")
except HTTPError as exc:
body = exc.read().decode("utf-8", errors="replace")
raise RuntimeError(f"CGI returned HTTP {exc.code}: {body[:500]}") from exc
except URLError as exc:
raise RuntimeError(f"Cannot reach CGI service {SERVICE_URL}: {exc.reason}") from exc
@mcp.resource("verbssql://info", mime_type="application/json")
def service_info_resource() -> str:
return json.dumps(
{
"service_url": SERVICE_URL,
"method": "POST",
"content_type": "application/x-www-form-urlencoded; charset=UTF-8",
"field": "sql",
"purpose": "Execute arbitrary read-only SQL against the remote Greek verbs SQLite database.",
},
ensure_ascii=False,
indent=2,
)
@mcp.tool()
def query_sql(sql: str) -> dict[str, Any]:
"""Execute a read-only SQL query through the remote verbssql CGI service.
Send a SELECT or WITH query exactly as SQL. The CGI service performs the
authoritative read-only validation. This MCP bridge additionally rejects
statements that do not begin with SELECT or WITH, then POSTs the SQL in the
form field 'sql' and converts the returned HTML table to structured data.
The MCP server does not contain a hard-coded database schema. If table or
column names are unknown, discover the current SQLite schema through this
same tool before composing the actual query, for example:
SELECT name
FROM sqlite_master
WHERE type='table'
ORDER BY name;
and for a specific table:
SELECT *
FROM pragma_table_info('poiw');
Prefer schema discovery over guessing table or column names.
"""
if not isinstance(sql, str) or not sql.strip():
raise ValueError("SQL query must not be empty")
if not READ_ONLY_START.match(sql):
raise ValueError("Only read-only SQL beginning with SELECT or WITH is allowed")
document = _post_sql(sql)
result = _parse_table(document)
result.update({"sql": sql, "service_url": SERVICE_URL})
return result
if __name__ == "__main__":
mcp.run(transport="stdio")
The final architecture: MCP server and SQLite reside together on www.drbreinlinger.de.
ProxyPreserveHost On
ProxyPass "/mcp/verbs" \
"http://127.0.0.1:8765/mcp"
ProxyPassReverse "/mcp/verbs" \
"http://127.0.0.1:8765/mcp"
User=mcp
Group=mcp
SupplementaryGroups=www-data
ExecStart=.../.venv/bin/python \
.../server.py
Restart=on-failure
.mcp.jsonNo local MCP server is required on the client anymore. The following project configuration is sufficient; the bearer token is supplied through VERBS_MCP_TOKEN.
{
"mcpServers": {
"greek-verbs": {
"type": "http",
"url": "https://www.drbreinlinger.de/mcp/verbs",
"headers": {
"Authorization": "Bearer ${VERBS_MCP_TOKEN}"
}
}
}
}
Projekt/.mcp.json
Environment:
VERBS_MCP_TOKEN=...
No local venv
No local DB
No local MCP process
/usr/local/lib/verbs-remote-mcp/
server.py
.venv/
/etc/systemd/system/
verbs-remote-mcp.service
Apache VirtualHost:
/mcp/verbs → 127.0.0.1:8765/mcp
/var/www/html/Public/Greek/
verbsdynamic.db
.mcp.json, and the access token remain on the client.#!/usr/bin/env python3
"""Remote MCP server for read-only access to verbsdynamic.db."""
from __future__ import annotations
import os
import re
import sqlite3
import time
from pathlib import Path
from typing import Any
from mcp.server import MCPServer
from mcp.server.transport_security import TransportSecuritySettings
SERVER_NAME = "greek-verbs-sql"
DB_PATH = Path(os.environ.get(
"VERBS_MCP_DB",
"/var/www/html/Public/Greek/verbsdynamic.db",
))
HOST = os.environ.get("VERBS_MCP_HOST", "127.0.0.1")
PORT = int(os.environ.get("VERBS_MCP_PORT", "8765"))
PUBLIC_HOST = os.environ.get("VERBS_MCP_PUBLIC_HOST", "www.drbreinlinger.de")
MAX_ROWS = int(os.environ.get("VERBS_MCP_MAX_ROWS", "500"))
MAX_SECONDS = float(os.environ.get("VERBS_MCP_MAX_SECONDS", "3.0"))
READ_ONLY_START = re.compile(r"^\s*(?:SELECT|WITH)\b", re.IGNORECASE)
mcp = MCPServer(
SERVER_NAME,
instructions=(
"This server exposes a read-only SQLite database through query_sql(). "
"The schema is deliberately not hard-coded. If table or column names "
"are unknown, discover them with SELECTs against sqlite_master and "
"table-valued pragma functions such as pragma_table_info(). "
"Prefer schema discovery over guessing identifiers."
),
)
# Read-only schema PRAGMAs that SQLite may invoke internally for table-valued
# pragma functions such as SELECT * FROM pragma_table_info('poiw').
SAFE_PRAGMAS = {
"table_info",
"table_xinfo",
"index_list",
"index_info",
"index_xinfo",
"foreign_key_list",
}
def _authorizer(
action: int,
arg1: str | None,
arg2: str | None,
dbname: str | None,
source: str | None,
) -> int:
allowed = {
sqlite3.SQLITE_SELECT,
sqlite3.SQLITE_READ,
sqlite3.SQLITE_FUNCTION,
}
if hasattr(sqlite3, "SQLITE_RECURSIVE"):
allowed.add(sqlite3.SQLITE_RECURSIVE)
if action in allowed:
return sqlite3.SQLITE_OK
if action == sqlite3.SQLITE_PRAGMA and (arg1 or "").lower() in SAFE_PRAGMAS:
return sqlite3.SQLITE_OK
return sqlite3.SQLITE_DENY
def _connect() -> sqlite3.Connection:
if not DB_PATH.is_file():
raise RuntimeError(f"SQLite database not found: {DB_PATH}")
con = sqlite3.connect(
f"file:{DB_PATH}?mode=ro",
uri=True,
timeout=2.0,
)
con.execute("PRAGMA query_only = ON")
con.set_authorizer(_authorizer)
deadline = time.monotonic() + MAX_SECONDS
def progress() -> int:
return 1 if time.monotonic() > deadline else 0
# Called every 1000 SQLite virtual-machine instructions.
con.set_progress_handler(progress, 1000)
return con
@mcp.tool()
def query_sql(sql: str) -> dict[str, Any]:
"""Execute one read-only SQLite SELECT/WITH query.
The database schema is not hard-coded into this MCP server. When tables or
columns are unknown, discover the current schema through this same tool.
Examples:
SELECT name
FROM sqlite_master
WHERE type='table'
ORDER BY name;
SELECT *
FROM pragma_table_info('poiw');
Then compose the actual query from the discovered identifiers. Prefer
schema discovery over guessing table or column names.
Only SELECT or WITH statements are accepted. SQLite itself is additionally
opened read-only, query_only is enabled, and an authorizer rejects writes
and unsafe PRAGMAs. Results are capped at MAX_ROWS and execution time is
bounded.
"""
if not isinstance(sql, str) or not sql.strip():
raise ValueError("SQL query must not be empty")
if not READ_ONLY_START.match(sql):
raise ValueError("Only read-only SQL beginning with SELECT or WITH is allowed")
try:
with _connect() as con:
cur = con.execute(sql)
columns = [d[0] for d in cur.description] if cur.description else []
rows = cur.fetchmany(MAX_ROWS + 1)
except sqlite3.OperationalError as exc:
if "interrupted" in str(exc).lower():
raise RuntimeError(
f"SQLite query exceeded the {MAX_SECONDS:g}s execution limit"
) from exc
raise
truncated = len(rows) > MAX_ROWS
if truncated:
rows = rows[:MAX_ROWS]
return {
"columns": columns,
"rows": [list(row) for row in rows],
"row_count": len(rows),
"truncated": truncated,
"max_rows": MAX_ROWS,
}
if __name__ == "__main__":
security = TransportSecuritySettings(
allowed_hosts=[
PUBLIC_HOST,
f"{PUBLIC_HOST}:*",
"127.0.0.1",
"127.0.0.1:*",
"localhost",
"localhost:*",
],
allowed_origins=[
f"https://{PUBLIC_HOST}",
],
)
mcp.run(
transport="streamable-http",
host=HOST,
port=PORT,
streamable_http_path="/mcp",
transport_security=security,
)
unzip verbs-remote-mcp-mcpuser.zip
cd verbs-remote-mcp-mcpuser
sudo ./install.sh
mcp
sudo -u mcp test -r \
/var/www/html/Public/Greek/verbsdynamic.db \
&& echo readable
sudo systemctl enable --now verbs-remote-mcp
systemctl status verbs-remote-mcp
sudo a2enmod proxy proxy_http
sudo apachectl configtest
sudo systemctl reload apache2
export VERBS_MCP_TOKEN='...'
claude mcp list
| Variant | MCP location | DB location | Transport | Schema | Intermediate layer |
|---|---|---|---|---|---|
| A · Local MCP / Local DB (generic) | Client | Client | stdio | fixed + resource | — |
| B · Local MCP / Remote DB (limited) | Client | Server | stdio + HTTPS | domain-specific | CGI + HTML |
| C · Local MCP / Remote DB (generic) | Client | Server | stdio + HTTPS | dynamic | CGI + HTML |
| D · Remote MCP / Remote DB (generic) | Server | Server | HTTPS / MCP | dynamic | Apache reverse proxy |