file:///home/app/config.yaml
logs://session-123/10
mysql://demo_db/users
postgres://warehouse/orders/schema

from typing import Any, Dict

@mcp.resource("logs://{session_id}/{limit}")
def get_query_logs(limit: str = "5", session_id: str = "anonymous") -> Dict[str, Any]:
    """读取指定会话最近的 SQL 查询日志。"""
    try:
        limit_val = int(limit)
        if limit_val <= 0:
            return {
                "success": False,
                "error": "limit must be a positive integer",
            }

        logs = query_logger.get_logs(
            session_id=session_id,
            limit=limit_val,
        )
        total = query_logger.total_query_count(session_id=session_id)

        return {
            "success": True,
            "logs": logs,
            "total_queries": total,
        }
    except Exception as exc:
        return {
            "success": False,
            "error": str(exc),
        }

from mcp.types import Tool

def as_tool(self) -> Tool:
    return Tool(
        name=self.name,
        description=self.description,
        inputSchema={
            "type": "object",
            "properties": {
                "host": {
                    "type": "string",
                    "description": "数据库主机地址",
                },
                "port": {
                    "type": "integer",
                    "description": "数据库端口",
                },
                "user": {
                    "type": "string",
                    "description": "数据库用户名",
                },
                "password": {
                    "type": "string",
                    "description": "数据库密码",
                },
                "database": {
                    "type": "string",
                    "description": "数据库名称",
                },
            },
            "required": ["host", "port", "user", "password", "database"],
        },
    )

from typing import Any, Dict
import pymysql

@mcp.tool()
def query_data(sql: str, session_id: str = "anonymous") -> Dict[str, Any]:
    """执行只读 SQL 查询。"""
    conn = get_connection()
    cursor = None

    try:
        cursor = conn.cursor(pymysql.cursors.DictCursor)

        cursor.execute("SET TRANSACTION READ ONLY")
        cursor.execute("START TRANSACTION")

        try:
            cursor.execute(sql)
            results = cursor.fetchall()
            conn.commit()

            log_query(
                operation=sql,
                success=True,
                session_id=session_id,
            )

            return {
                "success": True,
                "results": results,
                "rowCount": len(results),
            }
        except Exception as exc:
            conn.rollback()
            log_query(
                operation=sql,
                success=False,
                error=str(exc),
                session_id=session_id,
            )
            return {
                "success": False,
                "error": str(exc),
            }
    finally:
        if cursor:
            cursor.close()
        conn.close()

查一下最近一周失败订单数量最高的商品。

@mcp.prompt()
def generate_db_gpt_prompt() -> str:
    """生成数据库问答任务提示词。"""
    tables_info = get_tables()
    database_name = tables_info["database"]
    tables = tables_info["tables"]

    table_definitions = []
    for table in tables:
        table_desc = get_table_description(table)
        if table_desc.get("success"):
            table_definitions.append(table_desc["table_definition"])

    return DB_GPT_SYSTEM_PROMPT.format(
        database_name=database_name,
        table_definitions="\n".join(table_definitions),
    )

DB_GPT_SYSTEM_PROMPT = """
你是一个数据库查询助手，需要根据用户问题生成安全、正确的 MySQL 查询语句。

数据库名：
{database_name}

可用表结构：
{table_definitions}

约束：
1. 只能生成 DQL 查询语句，例如 SELECT。
2. 只能使用已提供的表和字段。
3. 查询结果最多限制为 10000 行。
4. 生成 SQL 后需要检查语法和字段是否存在。
5. 如果可以给出索引优化建议，需要结合表结构和查询条件说明原因。
6. 输出必须是 JSON。

用户问题：
{user_question}

返回格式：
{{
  "thoughts": "分析思路",
  "sql": "SQL 查询语句",
  "display_type": "Table 或 Text",
  "index_advice": "索引建议"
}}
"""

from contextlib import AsyncExitStack
from typing import Optional

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client


class MCPClient:
    def __init__(self):
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()

    async def connect_to_server(self, server_script_path: str):
        is_python = server_script_path.endswith(".py")
        is_js = server_script_path.endswith(".js")

        if not (is_python or is_js):
            raise ValueError("server script must be a .py or .js file")

        command = "python" if is_python else "node"

        server_params = StdioServerParameters(
            command=command,
            args=[server_script_path],
            env=None,
        )

        stdio_transport = await self.exit_stack.enter_async_context(
            stdio_client(server_params)
        )

        read_stream, write_stream = stdio_transport

        self.session = await self.exit_stack.enter_async_context(
            ClientSession(read_stream, write_stream)
        )

        await self.session.initialize()

        tools_response = await self.session.list_tools()
        print("tools:", [tool.name for tool in tools_response.tools])

        resources_response = await self.session.list_resources()
        print("resources:", [resource.uri for resource in resources_response.resources])

        prompts_response = await self.session.list_prompts()
        print("prompts:", [prompt.name for prompt in prompts_response.prompts])

import json

async def process_query(self, query: str) -> str:
    """使用大语言模型处理数据库相关问题。"""
    prompt_result = await self.session.get_prompt("generate_db_gpt_prompt")
    system_prompt = prompt_result.messages[0].content.text

    llm_response = self.llm.chat(
        system_prompt=system_prompt,
        content=query,
        response_format="json_object",
        conversation_history=self.conversation_history,
    )

    response_data = json.loads(llm_response)

    self.conversation_history.append({
        "role": "user",
        "content": query,
    })
    self.conversation_history.append({
        "role": "assistant",
        "content": llm_response,
    })

    if not response_data.get("sql"):
        return json.dumps(response_data, ensure_ascii=False, indent=2)

    query_result = await self.session.call_tool(
        "query_data",
        {
            "sql": response_data["sql"],
            "session_id": self.session_id,
        },
    )

    final_response = {
        "thoughts": response_data.get("thoughts"),
        "sql": response_data["sql"],
        "display_type": response_data.get("display_type", "Table"),
        "results": (
            json.loads(query_result.content[0].text)
            if query_result.content and query_result.content[0].text
            else None
        ),
    }

    return json.dumps(final_response, ensure_ascii=False, indent=2)

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "query_data",
    "arguments": {
      "sql": "SELECT * FROM users LIMIT 10",
      "session_id": "session-001"
    }
  }
}

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "content": [
      {
        "type": "text",
        "text": "{\"success\": true, \"rowCount\": 10}"
      }
    ]
  }
}

{
  "jsonrpc": "2.0",
  "id": 1,
  "error": {
    "code": -32602,
    "message": "Invalid params"
  }
}

json_payload = session_message.message.model_dump_json(
    by_alias=True,
    exclude_none=True,
)

await process.stdin.send((json_payload + "\n").encode("utf-8"))

line = await read_line()

try:
    message = types.JSONRPCMessage.model_validate_json(line)
except Exception as exc:
    await read_stream_writer.send(exc)
else:
    await read_stream_writer.send(SessionMessage(message))

import asyncio
from mcp.server.stdio import stdio_server

async def run_stdio():
    await initialize_global_resources()

    try:
        async with stdio_server() as (read_stream, write_stream):
            await app.run(
                read_stream,
                write_stream,
                app.create_initialization_options(),
            )
    finally:
        await close_global_resources()


if mode == "stdio":
    asyncio.run(run_stdio())

{
  "mcp_db": {
    "timeout": 60,
    "type": "stdio",
    "command": "uv",
    "args": [
      "--directory",
      "/path/to/mcp_for_db/src",
      "run",
      "-m",
      "server.mcp.server_mysql",
      "--mode",
      "stdio"
    ],
    "env": {
      "MYSQL_HOST": "localhost",
      "MYSQL_PORT": "3306",
      "MYSQL_USER": "root",
      "MYSQL_PASSWORD": "password",
      "MYSQL_DATABASE": "mcp_db",
      "MYSQL_ROLE": "admin",
      "PYTHONPATH": "/path/to/MCP-DB"
    }
  }
}

event: message
data: {"jsonrpc":"2.0","id":1,"result":{"content":[]}}

from starlette.applications import Starlette
from starlette.routing import Route, Mount
from starlette.responses import Response
import uvicorn

from mcp.server.sse import SseServerTransport

def run_sse():
    sse = SseServerTransport("/messages/")

    async def handle_sse(request):
        async with sse.connect_sse(
            request.scope,
            request.receive,
            request.send,
        ) as streams:
            await app.run(
                streams[0],
                streams[1],
                app.create_initialization_options(),
            )

        return Response(status_code=204)

    starlette_app = Starlette(
        routes=[
            Route("/sse", endpoint=handle_sse),
            Mount("/messages/", app=sse.handle_post_message),
        ],
    )

    config = uvicorn.Config(
        app=starlette_app,
        host="0.0.0.0",
        port=9000,
        loop="asyncio",
    )

    uvicorn.Server(config).run()

{
  "mysql_mcp_server": {
    "timeout": 60,
    "type": "sse",
    "url": "http://localhost:9000/sse"
  }
}

from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

async with streamablehttp_client("http://localhost:9000/mcp") as (
    read_stream,
    write_stream,
    _,
):
    async with ClientSession(read_stream, write_stream) as session:
        result = await session.initialize()
        tools = await session.list_tools()

curl -LsSf https://astral.sh/uv/install.sh | sh
source "$HOME/.local/bin/env"

uv --version

uv init MCP-DB
cd MCP-DB

uv add "mcp[cli]"

uv run mcp --help

# server.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("demo-server")


@mcp.tool()
def add(a: int, b: int) -> int:
    """计算两个整数之和。"""
    return a + b


@mcp.resource("config://app")
def get_config() -> str:
    """读取应用配置。"""
    return "env=dev\nversion=1.0.0"


@mcp.prompt()
def explain_prompt(topic: str) -> str:
    return f"请用工程师能理解的方式解释：{topic}"


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

uv run python server.py

DW-DBA-MCP/
├── datas/
│   ├── files/                  # 工具执行过的 SQL 文件
│   ├── logs/                   # 服务运行日志
│   └── version/
├── mcp_for_db/
│   ├── client/
│   │   ├── api.py              # FastAPI 接口
│   │   └── client.py           # 自建 MCP Client
│   ├── debug/
│   │   └── mcp_logger.py       # 记录 Client 与 Server 通信数据
│   ├── envs/
│   │   ├── common.env
│   │   ├── dify.env
│   │   └── mysql.env
│   ├── server/
│   │   ├── common/
│   │   │   ├── prompts.py      # 提示词模板
│   │   │   └── tools.py        # 工具描述
│   │   ├── core/
│   │   │   ├── base_server.py
│   │   │   ├── config_manager.py
│   │   │   ├── env_distribute.py
│   │   │   └── service_manager.py
│   │   ├── server_mysql/
│   │   │   ├── mysql_server.py
│   │   │   ├── prompts/
│   │   │   ├── resources/
│   │   │   └── tools/
│   │   ├── server_dify/
│   │   │   ├── dify_server.py
│   │   │   └── tools/
│   │   └── shared/
│   │       ├── oauth/
│   │       ├── security/       # SQL 鉴权
│   │       ├── templates/
│   │       └── utils/
├── pyproject.toml
├── requirements.txt
└── uv.lock

from typing import List
from pydantic import AnyUrl
from mcp.types import Resource

async def setup_server(self):
    """设置 MCP Server 路由。"""
    if self.server_setup_completed:
        self.logger.debug("server routes already configured")
        return

    @self.server.list_resources()
    async def handle_list_resources() -> List[Resource]:
        registry = self.get_resource_registry()
        if registry is None:
            return []

        if hasattr(registry, "get_all_resources"):
            return await registry.get_all_resources()

        return []

    @self.server.read_resource()
    async def handle_read_resource(uri: AnyUrl) -> str:
        registry = self.get_resource_registry()
        if registry is None:
            raise ValueError("resource registry is not initialized")

        content = await registry.get_resource(uri)
        return content if content is not None else "null"

from typing import ClassVar, Dict, List, Type
from urllib.parse import urlparse
from pydantic import AnyUrl
from mcp.types import Resource

class ResourceRegistry:
    """管理所有资源实例。"""

    _resources: ClassVar[Dict[str, "BaseResource"]] = {}

    @classmethod
    def register(cls, resource_class: Type["BaseResource"]):
        resource = resource_class()
        cls._resources[str(resource.uri)] = resource

    @classmethod
    def register_instance(cls, resource: "BaseResource"):
        cls._resources[str(resource.uri)] = resource

    @classmethod
    async def get_resource(cls, uri: AnyUrl) -> str:
        parsed = urlparse(str(uri))
        path_parts = parsed.path.strip("/").split("/")

        if not path_parts or not path_parts[0]:
            raise ValueError(f"invalid resource uri: {uri}")

        uri_str = f"{parsed.scheme}://{parsed.netloc}/{parsed.path}"

        for resource in cls._resources.values():
            if str(resource.uri) == uri_str:
                return await resource.read_resource(uri)

        for resource in cls._resources.values():
            if str(resource.uri).endswith(path_parts[0]):
                return await resource.read_resource(uri)

        raise ValueError(f"unregistered resource: {uri}")

    @classmethod
    async def get_all_resources(cls) -> List[Resource]:
        result = []

        # 避免遍历过程中动态注册表资源导致字典变化。
        resources_copy = list(cls._resources.values())

        for resource in resources_copy:
            descriptions = await resource.get_resource_descriptions()
            result.extend(descriptions)

        return result

from typing import List
from pydantic import AnyUrl
from mcp.types import Resource

class BaseResource:
    """资源基类。"""

    name: str = ""
    description: str = ""
    uri: AnyUrl | None = None
    mimeType: str = "text/plain"
    auto_register: bool = True

    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)

        if cls.auto_register and cls.uri is not None:
            ResourceRegistry.register(cls)

    async def get_resource_descriptions(self) -> List[Resource]:
        raise NotImplementedError

    async def read_resource(self, uri: AnyUrl) -> str:
        raise NotImplementedError

import aiomysql
from pydantic import AnyUrl

class TableResource(BaseResource):
    """代表具体数据库表。"""

    auto_register = False

    TABLE_EXISTS_QUERY = """
        SELECT COUNT(*) AS table_exists
        FROM information_schema.tables
        WHERE table_schema = %s AND table_name = %s
    """

    COLUMN_METADATA_QUERY = """
        SELECT COLUMN_NAME, DATA_TYPE
        FROM information_schema.columns
        WHERE table_schema = %s AND table_name = %s
        ORDER BY ORDINAL_POSITION
    """

    def __init__(self, db_name: str, table_name: str, description: str):
        self.db_name = db_name
        self.table_name = table_name
        self.name = f"table: {table_name}"
        self.uri = AnyUrl(f"mysql://{db_name}/{table_name}")
        self.description = description
        self.mimeType = "text/csv"

    async def get_resource_descriptions(self):
        return []

    async def read_resource(self, uri: AnyUrl) -> str:
        table_name = extract_table_name(uri)
        column_metadata = await self.get_table_metadata(table_name)

        async with get_current_database_manager().get_connection() as conn:
            async with conn.cursor(aiomysql.DictCursor) as cursor:
                safe_query = build_safe_select_query(table_name)
                await cursor.execute(safe_query)

                columns = [col[0] for col in cursor.description]
                rows = await cursor.fetchall()

                return generate_csv(columns, rows, column_metadata)

    async def get_table_metadata(self, table_name: str):
        db_name = get_current_database_manager().get_current_config()["database"]

        async with get_current_database_manager().get_connection() as conn:
            async with conn.cursor(aiomysql.DictCursor) as cursor:
                await cursor.execute(
                    self.TABLE_EXISTS_QUERY,
                    (db_name, table_name),
                )
                exists = await cursor.fetchone()

                if not exists or not exists["table_exists"]:
                    raise ValueError(
                        f"table {table_name!r} does not exist in database {db_name!r}"
                    )

                await cursor.execute(
                    self.COLUMN_METADATA_QUERY,
                    (db_name, table_name),
                )

                rows = await cursor.fetchall()
                return [
                    (row["COLUMN_NAME"], row["DATA_TYPE"])
                    for row in rows
                ]

from mcp.types import Resource
from pydantic import AnyUrl

class MySQLResource(BaseResource):
    """MySQL 数据库根资源。"""

    name = "MySQL 数据库"
    uri = AnyUrl("mysql://localhost/default")
    description = "提供 MySQL 数据库表访问能力"
    mimeType = "text/csv"
    auto_register = True

    TABLE_QUERY = """
        SELECT
            TABLE_NAME AS table_name,
            TABLE_COMMENT AS table_comment,
            TABLE_ROWS AS estimated_rows
        FROM information_schema.tables
        WHERE table_schema = %s
    """

    def __init__(self):
        self.cache = {}

    async def get_resource_descriptions(self):
        db_manager = get_current_database_manager()
        if db_manager is None:
            return []

        db_name = db_manager.get_current_config().get("database")
        if not db_name:
            return []

        if "table_descriptions" in self.cache:
            return self.cache["table_descriptions"]

        async with db_manager.get_connection() as conn:
            async with conn.cursor(aiomysql.DictCursor) as cursor:
                await cursor.execute(self.TABLE_QUERY, (db_name,))
                tables = await cursor.fetchall()

                resources = []

                for table in tables:
                    table_name = table["table_name"]
                    description = table["table_comment"] or f"{table_name} 表"

                    if table["estimated_rows"]:
                        description += f" (~{table['estimated_rows']} 行)"

                    table_resource = TableResource(
                        db_name=db_name,
                        table_name=table_name,
                        description=description,
                    )

                    ResourceRegistry.register_instance(table_resource)

                    resources.append(
                        Resource(
                            uri=table_resource.uri,
                            name=table_resource.name,
                            mimeType=table_resource.mimeType,
                            description=table_resource.description,
                        )
                    )

                self.cache["table_descriptions"] = resources
                return resources

    async def read_resource(self, uri: AnyUrl) -> str:
        return json.dumps(
            {
                "name": self.name,
                "uri": str(self.uri),
                "description": self.description,
                "type": "database_root",
            },
            ensure_ascii=False,
        )

from .db_resource import MySQLResource, TableResource

__all__ = [
    "MySQLResource",
    "TableResource",
]

1. 根据用户描述调用 get_table_name，查找候选表。
2. 调用 get_table_desc，获取表结构和字段说明。
3. 结合用户条件生成 SELECT SQL。
4. 调用 explain_sql，检查执行计划。
5. 调用 query_data，执行只读查询。
6. 返回 SQL、查询结果和必要的优化建议。

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()
mcp_client = MCPClient()


class QueryRequest(BaseModel):
    question: str
    user_id: str
    database: str | None = None


@app.post("/db/query")
async def query_database(req: QueryRequest):
    result = await mcp_client.process_query(
        query=req.question,
        user_id=req.user_id,
        database=req.database,
    )

    return {
        "success": True,
        "data": result,
    }