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

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
  

添加增删改方法

1. @mcp.tool()
2. def insert_data(table_name: str, data: Dict[str, Any]) -> str:
3.     """Insert data into a specified table.
5.     Args:
6.         table_name: Name of the table to insert data into.
7.         data: A dictionary where keys are column names and values are the data to insert.
9.     Returns:
10.         A message indicating success or failure.
11.     """
12.     if not DB_PATH.exists():
13.         raise FileNotFoundError(f"Messages database not found at: {DB_PATH}")
15.     with SQLiteConnection(DB_PATH) as conn:
16.         cursor = conn.cursor()
18.         try:
19.             columns = ', '.join(data.keys())
20.             placeholders = ', '.join(['?' for _ in data])
21.             sql = f"INSERT INTO {table_name} ({columns}) VALUES ({placeholders})"
22.             cursor.execute(sql, list(data.values()))
23.             conn.commit()  # Commit the transaction
24.             return f"Data inserted successfully into '{table_name}'."
26.         except sqlite3.Error as e:
27.             raise ValueError(f"SQLite error: {str(e)}")
30. @mcp.tool()
31. def update_data(table_name: str, updates: Dict[str, Any], condition: str) -> str:
32.     """Update data in a specified table.
34.     Args:
35.         table_name: Name of the table to update.
36.         updates: A dictionary where keys are column names and values are the new data.
37.         condition: The condition for which rows to update (e.g., "id = 1").
39.     Returns:
40.         A message indicating success or failure.
41.     """
42.     if not DB_PATH.exists():
43.         raise FileNotFoundError(f"Messages database not found at: {DB_PATH}")
45.     with SQLiteConnection(DB_PATH) as conn:
46.         cursor = conn.cursor()
48.         try:
49.             updates_string = ', '.join([f"{key} = ?" for key in updates.keys()])
50.             sql = f"UPDATE {table_name} SET {updates_string} WHERE {condition}"
51.             cursor.execute(sql, list(updates.values()))
52.             conn.commit()  # Commit the transaction
53.             return f"Data updated successfully in '{table_name}' where {condition}."
55.         except sqlite3.Error as e:
56.             raise ValueError(f"SQLite error: {str(e)}")
59. @mcp.tool()
60. def delete_data(table_name: str, condition: str) -> str:
61.     """Delete data from a specified table.
63.     Args:
64.         table_name: Name of the table to delete data from.
65.         condition: The condition for which rows to delete (e.g., "id = 1").
67.     Returns:
68.         A message indicating success or failure.
69.     """
70.     if not DB_PATH.exists():
71.         raise FileNotFoundError(f"Messages database not found at: {DB_PATH}")
73.     with SQLiteConnection(DB_PATH) as conn:
74.         cursor = conn.cursor()
76.         try:
77.             sql = f"DELETE FROM {table_name} WHERE {condition}"
78.             cursor.execute(sql)
79.             conn.commit()  # Commit the transaction
80.             return f"Data deleted successfully from '{table_name}' where {condition}."
82.         except sqlite3.Error as e:
83.             raise ValueError(f"SQLite error: {str(e)}")

复制代码

初始化数据库

1. sqlite3 c:\users\user\my.db

复制代码

编译项目文件

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

复制代码

在cursor里添加mcp server

1. {
2.     "mcpServers": {
3.       "sqlite-explorer": {
4.         "command": "uv",
5.         "args": [
6.           "run",
7.           "--with",
8.           "fastmcp",
9.           "--with",
10.           "uvicorn",
11.           "fastmcp",
12.           "run",
13.           "E:/github/sqlite-explorer-fastmcp-mcp-server/sqlite_explorer.py"
14.         ],
15.         "env": {
16.           "SQLITE_DB_PATH": "c:/users/user/my.db"
17.         }
18.       }
19.     }
20.   }

复制代码

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

在cursor的chat里就可以对话了

• 它在大模型(mcp client)翻译后会与你本地配置的mcp server通讯
  

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！