|
|
from mcp.server.fastmcp import FastMCP |
|
|
import sqlite3, json |
|
|
|
|
|
mcp = FastMCP("EnterpriseData") |
|
|
|
|
|
conn = sqlite3.connect(":memory:", check_same_thread=False) |
|
|
cur = conn.cursor() |
|
|
|
|
|
cur.execute(""" |
|
|
CREATE TABLE Customers ( |
|
|
CustomerID INTEGER PRIMARY KEY AUTOINCREMENT, |
|
|
Name TEXT, |
|
|
Region TEXT, |
|
|
LastOrderDate TEXT |
|
|
) |
|
|
""") |
|
|
cur.executemany( |
|
|
"INSERT INTO Customers (Name, Region, LastOrderDate) VALUES (?, ?, ?)", |
|
|
[ |
|
|
("Acme Corp", "Northeast", "2024-12-01"), |
|
|
("Beta Inc", "West", "2025-06-01"), |
|
|
("Gamma Co", "Northeast", "2023-09-15"), |
|
|
("Delta LLC", "South", "2025-03-20"), |
|
|
("Epsilon Ltd", "Northeast", "2025-07-10") |
|
|
] |
|
|
) |
|
|
conn.commit() |
|
|
|
|
|
@mcp.tool() |
|
|
def query_database(sql: str) -> str: |
|
|
"""Run a SQL query and return JSON.""" |
|
|
try: |
|
|
cur.execute(sql) |
|
|
cols = [d[0] for d in cur.description or []] |
|
|
rows = [dict(zip(cols, r)) for r in cur.fetchall()] |
|
|
return json.dumps(rows) |
|
|
except Exception as e: |
|
|
return json.dumps({"error": str(e)}) |
|
|
|
|
|
if __name__ == "__main__": |
|
|
mcp.run(transport="stdio") |
|
|
|
