开发一个mcp-server实现sqlite智能体

1个月前发布

mcp介绍

MCP（Model Context Protocol）是由 Anthropic（Claude的那个公司）推出的开放标准协议，它为开发者提供了一个强大的工具，能够在数据源和 AI 驱动工具之间建立安全的双向连接。

举个好理解的例子：如果把 AI 比作电脑主机，那么 MCP 就相当于 USB 协议，而 MCP Server 则类似于各种 USB 设备（如摄像头、麦克风等）。通过实现 MCP Server，我们可以让 AI 轻松连接到各种数据源，大大扩展其功能范围。添加 MCP Server 的作用简单的说就是让你的 AI 模型具备调用外部服务的能力。

https://modelcontextprotocol.io/introduction

下载python项目

项目本身只有读库，读表的方法，需要进行扩展
https://github.com/hannesrudolph/sqlite-explorer-fastmcp-mcp-server

添加增删改方法


@mcp.tool()
def insert_data(table_name: str, data: Dict[str, Any]) -> str:
    """Insert data into a specified table.

    Args:
        table_name: Name of the table to insert data into.
        data: A dictionary where keys are column names and values are the data to insert.

    Returns:
        A message indicating success or failure.
    """
    if not DB_PATH.exists():
        raise FileNotFoundError(f"Messages database not found at: {DB_PATH}")

    with SQLiteConnection(DB_PATH) as conn:
        cursor = conn.cursor()

        try:
            columns = ', '.join(data.keys())
            placeholders = ', '.join(['?' for _ in data])
            sql = f"INSERT INTO {table_name} ({columns}) VALUES ({placeholders})"
            cursor.execute(sql, list(data.values()))
            conn.commit()  # Commit the transaction
            return f"Data inserted successfully into '{table_name}'."

        except sqlite3.Error as e:
            raise ValueError(f"SQLite error: {str(e)}")


@mcp.tool()
def update_data(table_name: str, updates: Dict[str, Any], condition: str) -> str:
    """Update data in a specified table.

    Args:
        table_name: Name of the table to update.
        updates: A dictionary where keys are column names and values are the new data.
        condition: The condition for which rows to update (e.g., "id = 1").

    Returns:
        A message indicating success or failure.
    """
    if not DB_PATH.exists():
        raise FileNotFoundError(f"Messages database not found at: {DB_PATH}")

    with SQLiteConnection(DB_PATH) as conn:
        cursor = conn.cursor()

        try:
            updates_string = ', '.join([f"{key} = ?" for key in updates.keys()])
            sql = f"UPDATE {table_name} SET {updates_string} WHERE {condition}"
            cursor.execute(sql, list(updates.values()))
            conn.commit()  # Commit the transaction
            return f"Data updated successfully in '{table_name}' where {condition}."

        except sqlite3.Error as e:
            raise ValueError(f"SQLite error: {str(e)}")


@mcp.tool()
def delete_data(table_name: str, condition: str) -> str:
    """Delete data from a specified table.

    Args:
        table_name: Name of the table to delete data from.
        condition: The condition for which rows to delete (e.g., "id = 1").

    Returns:
        A message indicating success or failure.
    """
    if not DB_PATH.exists():
        raise FileNotFoundError(f"Messages database not found at: {DB_PATH}")

    with SQLiteConnection(DB_PATH) as conn:
        cursor = conn.cursor()

        try:
            sql = f"DELETE FROM {table_name} WHERE {condition}"
            cursor.execute(sql)
            conn.commit()  # Commit the transaction
            return f"Data deleted successfully from '{table_name}' where {condition}."

        except sqlite3.Error as e:
            raise ValueError(f"SQLite error: {str(e)}")

初始化数据库

sqlite3 c:\users\user\my.db

编译项目文件

uv run --with fastmcp --with uvicorn fastmcp run E:/github/sqlite-explorer-fastmcp-mcp-server/sqlite_explorer.py

在cursor里添加mcp server

{
    "mcpServers": {
      "sqlite-explorer": {
        "command": "uv",
        "args": [
          "run",
          "--with",
          "fastmcp",
          "--with",
          "uvicorn",
          "fastmcp",
          "run",
          "E:/github/sqlite-explorer-fastmcp-mcp-server/sqlite_explorer.py"
        ],
        "env": {
          "SQLITE_DB_PATH": "c:/users/user/my.db"
        }
      }
    }
  }

添加后，将它改为开启状态，绿色圆点