MCPDatabasesSSE
from mcp.server.fastmcp import FastMCP from connection import connect
mcp = FastMCP(name = "PostgressManagement", host="0.0.0.0", port=8050)
def execute(query: str): conn = connect() cursor = conn.cursor() cursor.execute(query) conn.commit() cursor.close() conn.close() return True
@mcp.tool() async def query_db(query: str): conn = connect() cursor = conn.cursor() cursor.execute(query) result = cursor.fetchall() cursor.close() conn.close() return result
@mcp.tool() async def insert_db(query: str): return execute(query)
@mcp.tool() async def update_db(query: str): return execute(query)
@mcp.tool() async def delete_db(query: str): return execute(query)
@mcp.tool() async def create_table(query: str): return execute(query)
@mcp.tool() async def alter_table(query: str): return execute(query)
@mcp.tool() async def drop_table(query: str): return execute(query)
@mcp.tool() async def insert_db(query: str): return execute(query)
if name == "main": print("Starting server...") conn = connect() mcp.run(transport="sse")
Server Config
{
"mcpServers": {
"SqliteManagement": {
"command": "/home/slendy/MCPProjects/DataBase/.venv/bin/python",
"args": [
"/home/slendy/MCPProjects/DataBase/main.py",
"--engine",
"sqlite",
"--url",
"sqlite:////home/slendy/MCPProjects/DataBase/ecommerce.db"
]
},
"PostgressManagement": {
"command": "/home/slendy/MCPProjects/DataBase/.venv/bin/python",
"args": [
"/home/slendy/MCPProjects/DataBase/main.py",
"--engine",
"postgresql",
"--url",
"postgresql://postgres:postgres@localhost:5433/ecommerce"
]
}
}
}