鑫金龙 發表於 2025-4-2 16:42:00

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

<h1 id="mcp介绍">mcp介绍</h1>
<p>MCP(Model Context Protocol)是由 Anthropic(Claude的那个公司) 推出的开放标准协议,它为开发者提供了一个强大的工具,能够在数据源和 AI 驱动工具之间建立安全的双向连接。</p>
<p>举个好理解的例子:如果把 AI 比作电脑主机,那么 MCP 就相当于 USB 协议,而 MCP Server 则类似于各种 USB 设备(如摄像头、麦克风等)。通过实现 MCP Server,我们可以让 AI 轻松连接到各种数据源,大大扩展其功能范围。添加 MCP Server 的作用简单的说就是让你的 AI 模型具备调用外部服务的能力。</p>
<p><img src="https://img2024.cnblogs.com/blog/118538/202504/118538-20250401132945587-1887880337.png" alt="" loading="lazy"></p>
<blockquote>
<p>https://modelcontextprotocol.io/introduction</p>
</blockquote>
<h1 id="下载python项目">下载python项目</h1>
<ul>
<li>项目本身只有读库,读表的方法,需要进行扩展</li>
<li>https://github.com/hannesrudolph/sqlite-explorer-fastmcp-mcp-server</li>
</ul>
<h1 id="添加增删改方法">添加增删改方法</h1>
<pre><code class="language-python">
@mcp.tool()
def insert_data(table_name: str, data: Dict) -&gt; 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, condition: str) -&gt; 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()
            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) -&gt; 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)}")
</code></pre>
<h1 id="初始化数据库">初始化数据库</h1>
<pre><code>sqlite3 c:\users\user\my.db
</code></pre>
<h1 id="编译项目文件">编译项目文件</h1>
<pre><code>uv run --with fastmcp --with uvicorn fastmcp run E:/github/sqlite-explorer-fastmcp-mcp-server/sqlite_explorer.py

</code></pre>
<h1 id="在cursor里添加mcp-server">在cursor里添加mcp server</h1>
<pre><code>{
    "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"
      }
      }
    }
}
</code></pre>
<p>添加后,将它改为开启状态,<code>绿色圆点</code><br>
<img src="https://images.cnblogs.com/cnblogs_com/lori/2430318/o_250402081830_mcp_sqlite.png" alt="" loading="lazy"></p>
<h1 id="在cursor的chat里就可以对话了">在cursor的chat里就可以对话了</h1>
<ul>
<li>它在大模型(mcp client)翻译后会与你本地配置的mcp server通讯</li>
</ul>
<p><img src="https://images.cnblogs.com/cnblogs_com/lori/2430318/o_250402083347_%E5%BE%AE%E4%BF%A1%E6%88%AA%E5%9B%BE_20250402163219.png" alt="" loading="lazy"></p>


</div>
<div id="MySignature" role="contentinfo">
    <p></p>
<div class="navgood">
<p>作者:仓储大叔,张占岭,<br>
荣誉:微软MVP<br>QQ:853066980</p>

<p><strong>支付宝扫一扫,为大叔打赏!</strong>
<br><img src="https://images.cnblogs.com/cnblogs_com/lori/237884/o_IMG_7144.JPG"></p>
</div><br><br>
来源:https://www.cnblogs.com/lori/p/18806245
頁: [1]
查看完整版本: 开发一个mcp-server实现sqlite智能体