File size: 2,076 Bytes
4af70f5 8d0f69f 4af70f5 8d0f69f 27de058 4af70f5 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 |
import os
import uvicorn
from mcp.server.fastmcp import FastMCP
from starlette.requests import Request
from starlette.responses import JSONResponse, Response
from langchain_community.utilities import SQLDatabase
from langchain_community.tools.sql_database.tool import QuerySQLCheckerTool
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
api_key=os.environ.get('OPENAI_API_KEY', None),
base_url=os.environ['OPENAI_BASE_URL'],
model='gpt-4o-mini',
temperature=0
)
# Create an MCP server
mcp = FastMCP("Credit Card Database Server")
credit_card_db = SQLDatabase.from_uri(r"sqlite:///data/ccms.db")
query_checker_tool = QuerySQLCheckerTool(db=credit_card_db, llm=llm)
@mcp.custom_route("/", methods=["GET"])
async def home(request: Request) -> Response:
return JSONResponse(
status_code=200,
content={
"message": "This is an MCP server that interfaces with a credit card database"
}
)
@mcp.tool()
def sql_db_list_tables():
"""
Returns a comma-separated list of table names in the database.
"""
return credit_card_db.get_usable_table_names()
@mcp.tool()
def sql_db_schema(table_names: list[str]) -> str:
"""
Input 'table_names_str' is a comma-separated string of table names.
Returns the DDL SQL schema for these tables.
"""
return credit_card_db.get_table_info(table_names)
@mcp.tool()
def sql_db_query_checker(query: str) -> str:
"""
Input 'query' is a SQL query string.
Checks if the query is valid.
If the query is valid, it returns the original query.
If the query is not valid, it returns the corrected query.
This tool is used to ensure the query is valid before executing it.
"""
return query_checker_tool.run(query)
@mcp.tool()
def sql_db_query(query: str) -> str:
"""
Input 'query' is a SQL query string.
Executes the query (SELECT only) and returns the result.
"""
return credit_card_db.run(query)
if __name__ == "__main__":
uvicorn.run(mcp.streamable_http_app, host="0.0.0.0", port=8000) |