下載Trae

• https://www.trae.com.cn/
  

uv 工具教程參考我的這篇文章

• 讓 Python 項目管理變簡單（uv 工具快速上手指南）

創建 uv 項目

uv init demo
cd demo

• 添加依賴項

uv add 'mcp[cli]'

main.py

import asyncio
import argparse
import sqlite3
import logging
from contextlib import closing
from pathlib import Path
from pydantic import AnyUrl
from typing import Anyfrom mcp.server import InitializationOptions
from mcp.server.lowlevel import Server, NotificationOptions
from mcp.server.stdio import stdio_server
import mcp.types as typeslogger = logging.getLogger('mcp_sqlite_server')
logger.info("Starting MCP SQLite Server")class SqliteDatabase:def __init__(self, db_path: str):self.db_path = str(Path(db_path).expanduser())Path(self.db_path).parent.mkdir(parents=True, exist_ok=True)self._init_database()self.insights: list[str] = []def _init_database(self):"""Initialize connection to the SQLite database"""logger.debug("Initializing database connection")with closing(sqlite3.connect(self.db_path)) as conn:conn.row_factory = sqlite3.Rowconn.close()def _synthesize_memo(self) -> str:"""Synthesizes business insights into a formatted memo"""logger.debug(f"Synthesizing memo with {len(self.insights)} insights")if not self.insights:return "No business insights have been discovered yet."insights = "\n".join(f"- {insight}" for insight in self.insights)memo = "📊 Business Intelligence Memo 📊\n\n"memo += "Key Insights Discovered:\n\n"memo += insightsif len(self.insights) > 1:memo += "\nSummary:\n"memo += f"Analysis has revealed {len(self.insights)} key business insights that suggest opportunities for strategic optimization and growth."logger.debug("Generated basic memo format")return memodef _execute_query(self, query: str, params: dict[str, Any] | None = None) -> list[dict[str, Any]]:"""Execute a SQL query and return results as a list of dictionaries"""logger.debug(f"Executing query: {query}")try:with closing(sqlite3.connect(self.db_path)) as conn:conn.row_factory = sqlite3.Rowwith closing(conn.cursor()) as cursor:if params:cursor.execute(query, params)else:cursor.execute(query)if query.strip().upper().startswith(('INSERT', 'UPDATE', 'DELETE', 'CREATE', 'DROP', 'ALTER')):conn.commit()affected = cursor.rowcountlogger.debug(f"Write query affected {affected} rows")return [{"affected_rows": affected}]results = [dict(row) for row in cursor.fetchall()]logger.debug(f"Read query returned {len(results)} rows")return resultsexcept Exception as e:logger.error(f"Database error executing query: {e}")raiseasync def main(db_path: str):logger.info(f"Starting SQLite MCP Server with DB path: {db_path}")db = SqliteDatabase(db_path)server = Server("sqlite-manager")logger.debug("Registering handlers")@server.list_resources()async def handle_list_resources() -> list[types.Resource]:logger.debug("Handling list_resources request")return [types.Resource(uri=AnyUrl("memo://insights"),name="Business Insights Memo",description="A living document of discovered business insights",mimeType="text/plain",)]@server.read_resource()async def handle_read_resource(uri: AnyUrl) -> str:logger.debug(f"Handling read_resource request for URI: {uri}")if uri.scheme != "memo":logger.error(f"Unsupported URI scheme: {uri.scheme}")raise ValueError(f"Unsupported URI scheme: {uri.scheme}")path = str(uri).replace("memo://", "")if not path or path != "insights":logger.error(f"Unknown resource path: {path}")raise ValueError(f"Unknown resource path: {path}")return db._synthesize_memo()@server.list_tools()async def handle_list_tools() -> list[types.Tool]:"""List available tools"""return [types.Tool(name="read_query",description="Execute a SELECT query on the SQLite database",inputSchema={"type": "object","properties": {"query": {"type": "string", "description": "SELECT SQL query to execute"},},"required": ["query"],},),types.Tool(name="write_query",description="Execute an INSERT, UPDATE, or DELETE query on the SQLite database",inputSchema={"type": "object","properties": {"query": {"type": "string", "description": "SQL query to execute"},},"required": ["query"],},),types.Tool(name="create_table",description="Create a new table in the SQLite database",inputSchema={"type": "object","properties": {"query": {"type": "string", "description": "CREATE TABLE SQL statement"},},"required": ["query"],},),types.Tool(name="list_tables",description="List all tables in the SQLite database",inputSchema={"type": "object","properties": {},},),types.Tool(name="describe_table",description="Get the schema information for a specific table",inputSchema={"type": "object","properties": {"table_name": {"type": "string", "description": "Name of the table to describe"},},"required": ["table_name"],},),types.Tool(name="append_insight",description="Add a business insight to the memo",inputSchema={"type": "object","properties": {"insight": {"type": "string", "description": "Business insight discovered from data analysis"},},"required": ["insight"],},),]@server.call_tool()async def handle_call_tool(name: str, arguments: dict[str, Any] | None) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:"""Handle tool execution requests"""try:if name == "list_tables":results = db._execute_query("SELECT name FROM sqlite_master WHERE type='table'")return [types.TextContent(type="text", text=str(results))]elif name == "describe_table":if not arguments or "table_name" not in arguments:raise ValueError("Missing table_name argument")results = db._execute_query(f"PRAGMA table_info({arguments['table_name']})")return [types.TextContent(type="text", text=str(results))]elif name == "append_insight":if not arguments or "insight" not in arguments:raise ValueError("Missing insight argument")db.insights.append(arguments["insight"])_ = db._synthesize_memo()# Notify clients that the memo resource has changedawait server.request_context.session.send_resource_updated(AnyUrl("memo://insights"))return [types.TextContent(type="text", text="Insight added to memo")]if not arguments:raise ValueError("Missing arguments")if name == "read_query":if not arguments["query"].strip().upper().startswith("SELECT"):raise ValueError("Only SELECT queries are allowed for read_query")results = db._execute_query(arguments["query"])return [types.TextContent(type="text", text=str(results))]elif name == "write_query":if arguments["query"].strip().upper().startswith("SELECT"):raise ValueError("SELECT queries are not allowed for write_query")results = db._execute_query(arguments["query"])return [types.TextContent(type="text", text=str(results))]elif name == "create_table":if not arguments["query"].strip().upper().startswith("CREATE TABLE"):raise ValueError("Only CREATE TABLE statements are allowed")db._execute_query(arguments["query"])return [types.TextContent(type="text", text="Table created successfully")]else:raise ValueError(f"Unknown tool: {name}")except sqlite3.Error as e:return [types.TextContent(type="text", text=f"Database error: {str(e)}")]except Exception as e:return [types.TextContent(type="text", text=f"Error: {str(e)}")]async with stdio_server() as (read_stream, write_stream):logger.info("Server running with stdio transport")await server.run(read_stream,write_stream,InitializationOptions(server_name="sqlite",server_version="0.1.0",capabilities=server.get_capabilities(notification_options=NotificationOptions(),experimental_capabilities={},),),)if __name__ == "__main__":parser = argparse.ArgumentParser(description='SQLite MCP Server')parser.add_argument('--db-path', default="./sqlite_mcp_server.db", help='Path to SQLite database file')args = parser.parse_args()asyncio.run(main(args.db_path))

Cherry Studio 添加 MCP 服務

• 下載地址：https://cherry-ai.com/

• MCP 參數

--directory
~/TraeProjects/demo
run
main.py
--db-path
~/TraeProjects/demo/test.db

申請 DeepSeek API

• 申請 API_KEY：https://platform.deepseek.com/usage

配置 DeepSeek API

調用 MCP 服務

Trae 添加 MCP 服務

• 下載Trae：https://www.trae.com.cn/

添加 MCP

{"mcpServers": {"sqlite-server": {"command": "uv","args": ["--directory","~/TraeProjects/demo","run","main.py","--db-path","~/TraeProjects/demo/test.db"]}}
}

創建智能體

使用智能體

調用 MCP 創建 demo 表

查詢 demo 表結構信息

demo 表插入 2 條測試數據

查詢 demo 表中的數據