Sqlite3 Mcpserver

SQLite3-MCP聊天-IMDB电影数据库查询系统

一个智能聊天界面，允许用户使用自然语言查询IMDB电影数据库。由 官方模型上下文协议（MCP）Python SDK 和 谷歌双子座2.0 Flash，此应用程序演示了如何构建将LLM连接到外部数据源的生产就绪AI系统。

🎯 项目概述

该项目实现了 完整的人工智能数据库查询系统 用户可以用简单的英语提问，人工智能会自动：

1. 动态发现数据库架构
2. 生成正确的SQL查询（包括JOIN）
3. 安全执行查询
4. 格式以用户友好的方式呈现

不需要SQL知识 -自然地提问！

✨ 主要特点

🎬 交互示例

简单模式发现

You: "What tables are in the database?"

AI internally:
1. Calls list_tables() → ["movies", "directors"]
2. Returns formatted response

AI: "The database contains two tables: `movies` and `directors`."

使用JOIN的复杂查询

You: "Who are the top 5 directors by number of movies?"

AI internally:
1. Calls list_tables() → ["movies", "directors"]
2. Calls describe_table("movies") → sees director_id column
3. Calls describe_table("directors") → sees id column
4. Generates SQL:
   SELECT d.name, COUNT(*) as count 
   FROM movies m 
   JOIN directors d ON m.director_id = d.id 
   GROUP BY d.id 
   ORDER BY count DESC 
   LIMIT 5
5. Calls run_select(sql) → gets results
6. Formats output

AI: "The top 5 directors by number of movies are:
     1. Steven Spielberg (27 movies)
     2. Woody Allen (21 movies)
     3. Clint Eastwood (20 movies)
     4. Martin Scorsese (20 movies)
     5. Spike Lee (16 movies)"

🏗️ 建筑

┌──────────────────────────────────────────────────────────────────┐
│                            USER                                   │
│              Browser: http://localhost:8000                       │
└──────────────────────────┬───────────────────────────────────────┘
                           │
                           │ User types:
                           │ "Who directed Inception?"
                           ↓
┌──────────────────────────────────────────────────────────────────┐
│                  FRONTEND (Vanilla JS + HTML/CSS)                 │
│ ┌──────────────────────────────────────────────────────────────┐ │
│ │  • Modern chat UI with message bubbles                       │ │
│ │  • localStorage for chat history persistence                 │ │
│ │  • Session restart detection via serverSessionId            │ │
│ │  • Typing indicators, error handling                         │ │
│ └──────────────────────────────────────────────────────────────┘ │
│                                                                    │
│  POST /api/chat                                                    │
│  {                                                                 │
│    "message": "Who directed Inception?",                          │
│    "history": [...previous messages...]                           │
│  }                                                                 │
└──────────────────────────┬───────────────────────────────────────┘
                           │
                           ↓
┌──────────────────────────────────────────────────────────────────┐
│             BACKEND (FastAPI - main.py)                           │
│ ┌──────────────────────────────────────────────────────────────┐ │
│ │  Startup (lifespan context manager):                         │ │
│ │  1. Load LLM_API_KEY from .env                              │ │
│ │  2. Configure Gemini API                                     │ │
│ │  3. Create StdioServerParameters                             │ │
│ │  4. Launch MCP server as subprocess via stdio_client        │ │
│ │  5. Establish ClientSession with AsyncExitStack             │ │
│ │  6. Initialize MCP session (handshake)                       │ │
│ │  7. Fetch available tools via list_tools()                   │ │
│ │  8. Store 8 tools globally: ping, db_path, list_tables,    │ │
│ │     describe_table, preview, run_select, run_exec, explain  │ │
│ └──────────────────────────────────────────────────────────────┘ │
│                                                                    │
│ ┌──────────────────────────────────────────────────────────────┐ │
│ │  /api/chat endpoint processing:                              │ │
│ │  1. Validate message and API key                             │ │
│ │  2. Convert MCP tools → Gemini function declarations        │ │
│ │     (JSON Schema → Gemini Schema format)                     │ │
│ │  3. Create GenerativeModel with tools + system prompt       │ │
│ │  4. Start chat with history (last 10 messages)              │ │
│ │  5. Send user message to Gemini                              │ │
│ │  6. Enter function calling loop (max 10 iterations):        │ │
│ │     a. Check all response.parts for function_call           │ │
│ │     b. If found: extract tool_name and arguments            │ │
│ │     c. Call MCP tool via mcp_session.call_tool()           │ │
│ │     d. Send result back to Gemini as function_response      │ │
│ │     e. Repeat until no more function calls                   │ │
│ │  7. Extract final text response                              │ │
│ │  8. Return to frontend                                       │ │
│ └──────────────────────────────────────────────────────────────┘ │
└──────────────────────────┬───────────────────────────────────────┘
                           │
                           ↓
┌──────────────────────────────────────────────────────────────────┐
│                  GEMINI 2.0 FLASH (Google AI)                     │
│ ┌──────────────────────────────────────────────────────────────┐ │
│ │  Receives:                                                    │ │
│ │  • User message: "Who directed Inception?"                   │ │
│ │  • Available functions: list_tables, describe_table, etc.   │ │
│ │  • System instruction: "You are an expert at querying      │ │
│ │    IMDB database. Use tools to discover schema. Never       │ │
│ │    show SQL code to user. Be proactive."                    │ │
│ │  • Chat history (for context)                                │ │
│ │                                                               │ │
│ │  Decides:                                                     │ │
│ │  "I need to query the database. First, let me check         │ │
│ │   the schema to understand the structure."                   │ │
│ │                                                               │ │
│ │  Returns function_call:                                       │ │
│ │  {                                                            │ │
│ │    "name": "describe_table",                                 │ │
│ │    "arguments": {"table": "movies"}                          │ │
│ │  }                                                            │ │
│ └──────────────────────────────────────────────────────────────┘ │
└──────────────────────────┬───────────────────────────────────────┘
                           │
                           │ function_call detected
                           ↓
┌──────────────────────────────────────────────────────────────────┐
│           MCP CLIENT (Official Python SDK - mcp package)          │
│ ┌──────────────────────────────────────────────────────────────┐ │
│ │  • ClientSession - Manages protocol communication           │ │
│ │  • stdio_client - stdio transport for subprocess            │ │
│ │  • StdioServerParameters - Server configuration             │ │
│ │  • AsyncExitStack - Proper async resource management        │ │
│ │                                                               │ │
│ │  await mcp_session.call_tool("describe_table",              │ │
│ │                               {"table": "movies"})           │ │
│ │                                                               │ │
│ │  Behind the scenes (SDK handles):                            │ │
│ │  1. Serialize to JSON-RPC format                             │ │
│ │  2. Write to server's stdin                                  │ │
│ │  3. Read from server's stdout                                │ │
│ │  4. Deserialize JSON-RPC response                            │ │
│ │  5. Extract tool result content                              │ │
│ └──────────────────────────────────────────────────────────────┘ │
└──────────────────────────┬───────────────────────────────────────┘
                           │
                           │ JSON-RPC over stdin/stdout
                           │ (stdio transport)
                           ↓
┌──────────────────────────────────────────────────────────────────┐
│         MCP SERVER (server.py - Built with FastMCP)               │
│ ┌──────────────────────────────────────────────────────────────┐ │
│ │  8 MCP Tools (decorated with @app.tool()):                  │ │
│ │                                                               │ │
│ │  1. ping() → "pong"                                          │ │
│ │     Health check                                              │ │
│ │                                                               │ │
│ │  2. db_path() → "/path/to/movies.sqlite"                    │ │
│ │     Returns database file path                               │ │
│ │                                                               │ │
│ │  3. list_tables() → ["movies", "directors"]                 │ │
│ │     Lists all tables in database                             │ │
│ │                                                               │ │
│ │  4. describe_table(table) → [column info...]                │ │
│ │     Returns schema for a specific table                      │ │
│ │     (name, type, notnull, default, primary_key)             │ │
│ │                                                               │ │
│ │  5. preview(table, limit, offset) → [rows...]               │ │
│ │     Preview data from a table                                │ │
│ │                                                               │ │
│ │  6. run_select(sql, params, limit) → [rows...]              │ │
│ │     Execute SELECT query (read-only)                         │ │
│ │     Validates SQL starts with SELECT                         │ │
│ │                                                               │ │
│ │  7. run_exec(sql, params) → {rows_affected, last_row_id}   │ │
│ │     Execute write operations (INSERT/UPDATE/DELETE)          │ │
│ │                                                               │ │
│ │  8. explain(sql, params) → [query plan...]                  │ │
│ │     Returns SQLite query execution plan                      │ │
│ └──────────────────────────────────────────────────────────────┘ │
│                                                                    │
│  Each tool:                                                        │
│  • Opens SQLite connection (read-only by default)                 │
│  • Executes operation                                              │
│  • Returns result as JSON-serializable data                       │
│  • Handles errors                                                  │
└──────────────────────────┬───────────────────────────────────────┘
                           │
                           │ SQL Query Execution
                           ↓
┌──────────────────────────────────────────────────────────────────┐
│                  SQLITE DATABASE (movies.sqlite)                  │
│ ┌──────────────────────────────────────────────────────────────┐ │
│ │  Table: movies                                                │ │
│ │  ┌────────────────┬──────────┬─────────────────────────────┐ │ │
│ │  │ Column         │ Type     │ Description                 │ │ │
│ │  ├────────────────┼──────────┼─────────────────────────────┤ │ │
│ │  │ id             │ INTEGER  │ Primary key                 │ │ │
│ │  │ title          │ TEXT     │ Movie title                 │ │ │
│ │  │ original_title │ VARCHAR  │ Original title              │ │ │
│ │  │ release_date   │ TEXT     │ Release date                │ │ │
│ │  │ budget         │ INTEGER  │ Budget in dollars           │ │ │
│ │  │ revenue        │ INTEGER  │ Revenue in dollars          │ │ │
│ │  │ popularity     │ INTEGER  │ Popularity score            │ │ │
│ │  │ vote_average   │ REAL     │ Average rating              │ │ │
│ │  │ vote_count     │ INTEGER  │ Number of votes             │ │ │
│ │  │ overview       │ TEXT     │ Movie description           │ │ │
│ │  │ tagline        │ TEXT     │ Movie tagline               │ │ │
│ │  │ director_id    │ INTEGER  │ FK → directors.id           │ │ │
│ │  │ uid            │ INTEGER  │ Unique identifier           │ │ │
│ │  └────────────────┴──────────┴─────────────────────────────┘ │ │
│ │                                                                │ │
│ │  Table: directors                                              │ │
│ │  ┌────────────┬──────────┬──────────────────────────────────┐ │ │
│ │  │ Column     │ Type     │ Description                      │ │ │
│ │  ├────────────┼──────────┼──────────────────────────────────┤ │ │
│ │  │ id         │ INTEGER  │ Primary key                      │ │ │
│ │  │ name       │ TEXT     │ Director name                    │ │ │
│ │  │ gender     │ INTEGER  │ Gender (0=female, 1=male, etc.) │ │ │
│ │  │ uid        │ INTEGER  │ Unique identifier                │ │ │
│ │  │ department │ TEXT     │ Department (usually "Directing") │ │ │
│ │  └────────────┴──────────┴──────────────────────────────────┘ │ │
│ │                                                                │ │
│ │  Relationship: movies.director_id → directors.id (JOIN)       │ │
│ └──────────────────────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────────┘

🔧 完整的生命周期说明

第一阶段：服务器启动

当你奔跑时 uv run main.py，具体情况如下：

步骤1.1:FastAPI初始化

# main.py (lines 1-22)
load_dotenv()  # Load .env file
PROJECT_ROOT = Path(__file__).parent
MCP_SERVER_PATH = PROJECT_ROOT / "src" / "sqlite3_mcp" / "server.py"
LLM_API_KEY = os.environ.get('LLM_API_KEY', None)
SERVER_SESSION_ID = str(int(time.time()))  # Unix timestamp for restart detection

app = FastAPI(title="SQLite3 MCP Chat", lifespan=lifespan)

步骤1.2：生命周期上下文管理器执行

# main.py (lines 30-79)
@asynccontextmanager
async def lifespan(app: FastAPI):
    global mcp_session, mcp_tools_list, mcp_exit_stack
    
    # Configure Gemini with API key
    genai.configure(api_key=LLM_API_KEY)
    
    # Define how to start the MCP server subprocess
    server_params = StdioServerParameters(
        command="python",
        args=[str(MCP_SERVER_PATH)]
    )
    
    # Use AsyncExitStack for proper async context management
    mcp_exit_stack = AsyncExitStack()
    
    # Connect to MCP server via stdio (stdin/stdout)
    read, write = await mcp_exit_stack.enter_async_context(
        stdio_client(server_params)
    )
    
    # Create MCP client session
    mcp_session = await mcp_exit_stack.enter_async_context(
        ClientSession(read, write)
    )
    
    # MCP handshake - establish protocol version, capabilities
    await mcp_session.initialize()
    
    # Fetch all available tools from the server
    tools_response = await mcp_session.list_tools()
    mcp_tools_list = tools_response.tools  # List of 8 tools
    
    print(f"✅ MCP session initialized with {len(mcp_tools_list)} tools")
    
    yield  # FastAPI server runs here
    
    # Cleanup on shutdown (Ctrl+C)
    await mcp_exit_stack.aclose()

幕后发生了什么：

1. stdio_client(server_params) 生成一个子流程： python server.py
2. 子进程运行MCP服务器，监听stdin/stdout
3. ClientSession(read, write) 建立双向通信
4. initialize() 发送以下JSON-RPC消息：

   {
     "jsonrpc": "2.0",
     "method": "initialize",
     "params": {
       "protocolVersion": "2024-11-05",
       "capabilities": {},
       "clientInfo": {"name": "mcp-client", "version": "1.0.0"}
     }
   }

1. list_tools() 发送：

   {
     "jsonrpc": "2.0",
     "method": "tools/list",
     "params": {}
   }

1. 服务器返回8个工具定义

步骤1.3：服务器就绪

✅ MCP session initialized with 8 tools
   - ping: Quick health check
   - db_path: Return database path
   - list_tables: List all user tables
   - describe_table: Describe table columns
   - preview: Preview table rows
   - run_select: Execute SELECT query
   - run_exec: Execute write statement
   - explain: Query execution plan
✅ Gemini API configured (Pure MCP with Official SDK!)
INFO:     Uvicorn running on http://0.0.0.0:8000

第二阶段：用户交互

步骤2.1：前端初始化

// script.js (lines 26-37)
async function init() {
    setupEventListeners();  // Bind click handlers
    adjustTextareaHeight(); // Set textarea height
    
    // Check if server restarted by comparing session IDs
    const wasRestarted = await checkServerRestart();
    
    // Only load old chat if server didn't restart
    if (!wasRestarted) {
        loadChatHistory();  // Load from localStorage
    }
}

步骤2.2：会话重启检测

// script.js (lines 40-98)
async function checkServerRestart() {
    // Fetch current server session ID
    const response = await fetch('/api/status');
    const data = await response.json();
    const serverSessionId = data.session_id;  // e.g., "1762468389"
    
    // Get previously stored session ID
    const storedSessionId = localStorage.getItem('serverSessionId');
    
    // If they don't match, server was restarted
    if (storedSessionId && storedSessionId !== serverSessionId) {
        // Clear localStorage
        state.messageHistory = [];
        localStorage.removeItem('chatHistory');
        
        // Clear UI (remove all messages except welcome)
        const messages = elements.chatMessages.querySelectorAll('.message, .error-message');
        messages.forEach(msg => msg.remove());
        
        // Show restart notification
        showRestartNotification();
        
        // Store new session ID
        localStorage.setItem('serverSessionId', serverSessionId);
        
        return true;  // Restart detected
    }
    
    // Store session ID for next check
    localStorage.setItem('serverSessionId', serverSessionId);
    return false;  // No restart
}

为什么这很重要：

• 服务器重启=新MCP会话=新状态
• 旧的聊天历史记录可能会引用过时的工具调用
• 清除历史记录可防止混淆

步骤2.3：用户发送消息

User types: "Who are the top 5 directors by number of movies?"
Clicks send button (or presses Enter)

// script.js (lines 151-211)
async function handleSendMessage() {
    const message = elements.messageInput.value.trim();
    
    // Check for server restart before sending
    await checkServerRestart();
    
    // Clear input and show user message in UI
    elements.messageInput.value = '';
    addMessage(message, 'user');
    
    // Show typing indicator
    const typingId = showTypingIndicator();
    
    // Send to backend
    const response = await sendToLLM(message);
    
    // Remove typing indicator
    removeTypingIndicator(typingId);
    
    // Show AI response
    addMessage(response, 'assistant');
}

// script.js (lines 313-338)
async function sendToLLM(message) {
    const response = await fetch('/api/chat', {
        method: 'POST',
        headers: {'Content-Type': 'application/json'},
        body: JSON.stringify({
            message: message,
            history: state.messageHistory  // Last 10 messages
        })
    });
    
    const data = await response.json();
    return data.response;
}

阶段3：后端处理

步骤3.1：接收请求

# main.py (lines 223-242)
@app.post("/api/chat", response_model=MessageResponse)
async def chat(request: MessageRequest):
    message = request.message  # "Who are the top 5 directors..."
    
    # Validate
    if not message:
        raise HTTPException(400, "Message is required")
    if not LLM_API_KEY:
        raise HTTPException(500, "LLM API key not configured")
    if not mcp_session:
        raise HTTPException(500, "MCP session not initialized")

步骤3.2：将MCP工具转换为Gemini格式

# main.py (lines 85-141)
def convert_mcp_tools_to_gemini_format():
    gemini_tools = []
    
    for tool in mcp_tools_list:  # 8 tools from MCP server
        parameters = {
            "type_": "OBJECT",  # Gemini requires type_ with UPPERCASE
            "properties": {},
        }
        
        # Convert each parameter from MCP JSON Schema to Gemini Schema
        if tool.inputSchema:
            for prop_name, prop_schema in tool.inputSchema["properties"].items():
                prop_type = prop_schema.get("type", "string")
                
                # MCP: "string" → Gemini: "STRING"
                type_mapping = {
                    "string": "STRING",
                    "integer": "INTEGER",
                    "number": "NUMBER",
                    "boolean": "BOOLEAN",
                    "array": "ARRAY",
                    "object": "OBJECT"
                }
                
                gemini_prop = {
                    "type_": type_mapping.get(prop_type.lower(), "STRING"),
                    "description": prop_schema.get("description", "")
                }
                
                parameters["properties"][prop_name] = gemini_prop
        
        function_declaration = {
            "name": tool.name,
            "description": tool.description,
            "parameters": parameters
        }
        
        gemini_tools.append(function_declaration)
    
    return gemini_tools

转换示例：

# MCP Tool Definition (from server.py)
{
  "name": "describe_table",
  "description": "Describe a table's columns",
  "inputSchema": {
    "type": "object",
    "properties": {
      "table": {
        "type": "string",
        "description": "Table name"
      }
    },
    "required": ["table"]
  }
}

# Gemini Function Declaration (after conversion)
{
  "name": "describe_table",
  "description": "Describe a table's columns",
  "parameters": {
    "type_": "OBJECT",
    "properties": {
      "table": {
        "type_": "STRING",
        "description": "Table name"
      }
    },
    "required": ["table"]
  }
}

步骤3.3：使用工具创建Gemini模型

# main.py (lines 244-264)
model = genai.GenerativeModel(
    model_name='gemini-2.0-flash',
    tools=gemini_tools,  # Our 8 MCP tools converted to Gemini format
    system_instruction="""You are an expert at answering questions about an IMDB movies database.

You have access to MCP tools that can discover the database schema and execute queries.

Important: The movies table has a director_id foreign key that links to the directors table's id.

When answering questions:
1. Use the tools to discover the schema if needed
2. Generate and execute correct SQL queries  
3. Present results in a clear, user-friendly format
4. Never show SQL code to the user - only show the results

Be proactive - use the tools immediately without asking for permission."""
)

为什么这个系统会提示？

1. 领域知识：告诉人工智能 movies.director_id → directors.id 关系（无法通过工具发现）
2. 行为：“积极主动”=不要征求许可，只使用工具
3. 输出格式：“从不显示SQL”=仅提供用户友好的响应
4. 工作流程：指导工具使用（发现→ 怎么翻译→ 格式）

步骤3.4:开始与历史记录聊天

# main.py (lines 266-272)
chat_history = []
for msg in request.history[-10:]:  # Last 10 messages only
    role = "user" if msg["role"] == "user" else "model"
    chat_history.append({"role": role, "parts": [msg["text"]]})

chat = model.start_chat(history=chat_history)

步骤3.5：向双子座发送消息

# main.py (line 275)
response = chat.send_message(message)
# Sends: "Who are the top 5 directors by number of movies?"

第四阶段：Gemini处理

双子座收到：

• 用户留言：“电影数量排名前五的导演是谁？”
• 可用功能：8个MCP工具
• 系统说明：主动使用工具，不显示SQL
• 聊天记录：以前的消息上下文

Gemini分析并决定：

"To answer this question, I need to:
1. Understand the database structure
2. Query the data with proper JOIN
3. Format the results

Let me start by checking the schema."

Gemini返回a 函数调用 （非文本）：

response.candidates[0].content.parts[0].function_call = {
    "name": "describe_table",
    "args": {"table": "movies"}
}

阶段5：函数调用循环

步骤5.1：检测函数调用

# main.py (lines 277-293)
while iteration  str:
    print(f"🔵 Calling MCP tool: {tool_name}({arguments})")
    
    # Official SDK handles all JSON-RPC communication
    result = await mcp_session.call_tool(tool_name, arguments)
    
    # Extract text content from result
    if result.content:
        content_parts = []
        for content in result.content:
            if hasattr(content, 'text'):
                content_parts.append(content.text)
            elif hasattr(content, 'data'):
                content_parts.append(str(content.data))
        
        result_text = "\n".join(content_parts)
        print(f"✅ MCP tool result: {result_text[:100]}...")
        return result_text
    
    return str(result)

幕后（SDK处理此问题）：

// Client → Server (stdin)
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "describe_table",
    "arguments": {"table": "movies"}
  }
}

// Server → Client (stdout)
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "content": [{
      "type": "text",
      "text": "[{\"cid\": 0, \"name\": \"id\", \"type\": \"INTEGER\", ...}]"
    }]
  }
}

步骤5.4:MCP服务器执行工具

# server.py (lines 78-96)
@app.tool()
def describe_table(table: str) -> List[Dict[str, Any]]:
    """Describe a table's columns"""
    _ensure_db_exists()
    
    # Open read-only SQLite connection
    with _connect(readonly=True) as conn:
        # SQLite PRAGMA for table schema
        rows = conn.execute(f"PRAGMA table_info({table})").fetchall()
    
    # Format as list of dicts
    return [
        {
            "cid": r["cid"],
            "name": r["name"],
            "type": r["type"],
            "notnull": bool(r["notnull"]),
            "default_value": r["dflt_value"],
            "primary_key": bool(r["pk"]),
        }
        for r in rows
    ]

SQLite连接：

# server.py (lines 24-36)
def _connect(readonly: bool = True) -> sqlite3.Connection:
    if readonly:
        # Read-only URI connection (security!)
        uri = f"file:{DB_PATH.as_posix()}?mode=ro"
        conn = sqlite3.connect(uri, uri=True, check_same_thread=False)
    else:
        conn = sqlite3.connect(DB_PATH.as_posix(), check_same_thread=False)
    
    conn.row_factory = sqlite3.Row  # Dict-like row access
    return conn

步骤5.5：将结果发送回Gemini

# main.py (lines 304-316)
tool_result = await call_mcp_tool(tool_name, tool_args)
# Returns: "[{\"name\": \"id\", \"type\": \"INTEGER\", ...}]"

# Send function response back to Gemini
response = chat.send_message(
    genai.protos.Content(
        parts=[genai.protos.Part(
            function_response=genai.protos.FunctionResponse(
                name=tool_name,
                response={'result': tool_result}
            )
        )]
    )
)

步骤5.6：双子座决定下一步行动

双子座现在知道了 movies 表架构。它决定：

"Good! I see director_id column. Now let me check the directors table."

返回另一个 函数调用:

{
    "name": "describe_table",
    "args": {"table": "directors"}
}

循环继续 （迭代2）：

• 呼叫 describe_table("directors")
• 获取控制器架构
• 发送回Gemini

Gemini现在拥有这两种模式。它决定：

"Perfect! I can now write a SQL query with JOIN.
movies.director_id → directors.id
Need to COUNT grouped by director, ORDER BY count, LIMIT 5."

返回另一个 函数调用:

{
    "name": "run_select",
    "args": {
        "sql": "SELECT d.name, COUNT(*) as movie_count FROM movies m JOIN directors d ON m.director_id = d.id GROUP BY d.id, d.name ORDER BY movie_count DESC LIMIT 5",
        "limit": 100
    }
}

循环继续 （迭代3）：

• 呼叫 run_select(sql, limit=100)
• 在SQLite中执行查询
• 获取结果： [{"name": "Steven Spielberg", "movie_count": 27}, ...]
• 发送回Gemini

双子座现在有了数据。它决定：

"Great! I have the results. Now I can format them nicely for the user."

退货 文本 （无函数调用）：

response.text = "The top 5 directors by number of movies are:
1. Steven Spielberg (27 movies)
2. Woody Allen (21 movies)
3. Clint Eastwood (20 movies)
4. Martin Scorsese (20 movies)
5. Spike Lee (16 movies)"

步骤5.7：退出循环

# main.py (lines 281-293)
for part in response.candidates[0].content.parts:
    if part.function_call:
        has_function_call = True
        break

if not has_function_call:
    break  # EXIT LOOP - no more function calls!

步骤5.8：提取最终响应

# main.py (lines 318-320)
response_text = response.text  # Safe now - no function_call parts!
print(f"✅ Final response after {iteration} iterations: {response_text}\n")

return MessageResponse(
    response=response_text,
    status="success"
)

第六阶段：返回前端

后端返回JSON：

{
  "response": "The top 5 directors by number of movies are:\n1. Steven Spielberg (27 movies)\n2. Woody Allen (21 movies)\n3. Clint Eastwood (20 movies)\n4. Martin Scorsese (20 movies)\n5. Spike Lee (16 movies)",
  "status": "success"
}

前端接收并显示：

// script.js (lines 183-189)
const response = await sendToLLM(message);
removeTypingIndicator(typingId);
addMessage(response, 'assistant');  // Show in UI
updateStatus('success', 'Ready');

第7阶段：聊天记录持久化

// script.js (lines 242-247)
state.messageHistory.push({ role, text, timestamp: Date.now() });
saveChatHistory();

function saveChatHistory() {
    localStorage.setItem('chatHistory', JSON.stringify(state.messageHistory));
}

记录完整的交互：

state.messageHistory = [
  {
    "role": "user",
    "text": "Who are the top 5 directors by number of movies?",
    "timestamp": 1762468389000
  },
  {
    "role": "assistant",
    "text": "The top 5 directors by number of movies are:\n1. Steven Spielberg (27 movies)...",
    "timestamp": 1762468392000
  }
]

📁 项目结构

sqlite3-mcp/
├── main.py                          # FastAPI backend + Gemini + MCP client
│   ├── lifespan()                   # Startup/shutdown management
│   ├── convert_mcp_tools_to_gemini_format()
│   ├── call_mcp_tool()              # MCP SDK wrapper
│   ├── /api/chat                    # Main chat endpoint
│   ├── /api/status                  # Health check + session ID
│   └── Static file serving          # HTML/CSS/JS
│
├── src/sqlite3_mcp/
│   └── server.py                    # MCP Server (FastMCP)
│       ├── 8 @app.tool() decorators # MCP tool definitions
│       ├── _connect()               # SQLite connection manager
│       └── _ensure_db_exists()      # Validation
│
├── frontend/
│   ├── index.html                   # Chat UI structure
│   ├── style.css                    # Beautiful gradients, animations
│   └── script.js                    # State management, API calls
│       ├── init()                   # Initialization
│       ├── checkServerRestart()     # Session detection
│       ├── handleSendMessage()      # Message handling
│       ├── sendToLLM()              # API communication
│       └── Chat history management  # localStorage
│
├── movies.sqlite                    # IMDB database
│   ├── movies table (13 columns)
│   └── directors table (5 columns)
│
├── pyproject.toml                   # Dependencies
│   ├── mcp>=1.20.0                 # Official MCP SDK ⭐
│   ├── fastapi>=0.121.0
│   ├── uvicorn>=0.38.0
│   ├── google-generativeai>=0.8.0   # Gemini SDK
│   └── python-dotenv>=1.0.0
│
├── .env                             # Environment variables
│   ├── LLM_API_KEY=
│   └── PORT=8000
│
├── .gitignore                       # Git ignore rules
└── README.md                        # This file

🔑 技术栈深潜

1.模型上下文协议（MCP）-官方SDK

包裹： mcp>=1.20.0

模型上下文协议标准的官方Python实现。

我们使用什么：

• mcp.ClientSession -管理MCP协议通信
• mcp.client.stdio.stdio_client -用于子进程通信的stdio传输
• mcp.StdioServerParameters -服务器启动配置
• contextlib.AsyncExitStack -正确的异步资源管理

为什么是官方SDK？

了解更多：

• https://github.com/modelcontextprotocol/python-sdk
• https://spec.modelcontextprotocol.io/

2.谷歌双子座2.0 Flash

包裹： google-generativeai>=0.8.0

谷歌最新的快速LLM，支持本机函数调用。

我们使用什么：

• genai.GenerativeModel -带函数调用的模型
• genai.protos.FunctionDeclaration -工具定义
• genai.protos.Content / Part -消息结构
• genai.protos.FunctionResponse -工具结果格式

为什么使用Gemini的原生函数调用？

• ✅ 无框架开销（无LangChain！）
• ✅ 直接控制函数调用循环
• ✅ 快速响应时间（每次呼叫约500-1000ms）
• ✅ 与清晰的API进行简单集成

函数调用流程：

# 1. Create model with tools
model = genai.GenerativeModel(model_name='gemini-2.0-flash', tools=tools)

# 2. Send message
response = chat.send_message("Who directed Inception?")

# 3. Check for function call
if response.candidates[0].content.parts[0].function_call:
    function_call = response.candidates[0].content.parts[0].function_call
    
    # 4. Execute tool
    result = execute_tool(function_call.name, function_call.args)
    
    # 5. Send result back
    response = chat.send_message(function_response=result)

# 6. Get final text
final_answer = response.text

了解更多： https://ai.google.dev/gemini-api/docs/function-calling

3.FastAPI

包裹： fastapi>=0.121.0 + uvicorn>=0.38.0

支持异步的现代Python web框架。

我们使用什么：

• @asynccontextmanager -启动/关闭的寿命事件
• @app.get/post -路线装饰师
• BaseModel -请求/响应的Pydantic模型
• FileResponse -静态文件服务
• HTTPException -错误处理

为什么选择FastAPI？

• ✅ 原生异步支持（非常适合MCP/GGemini调用）
• ✅ API自动文档（Swagger UI）
• ✅ Pydantic的类型安全
• ✅ 快速性能（Starlette+uvicorn）

寿命模式（现代FastAPI）：

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup code
    print("Starting up...")
    yield
    # Shutdown code
    print("Shutting down...")

app = FastAPI(lifespan=lifespan)

旧模式（已弃用）：

@app.on_event("startup")  # ❌ Deprecated!
async def startup():
    ...

4.FastMCP（MCP服务器SDK）

包裹： mcp>=1.20.0 （包括FastMCP）

内置于MCP Python SDK中，用于创建服务器。

我们使用什么：

from mcp.server.fastmcp import FastMCP

app = FastMCP("sqlite3-mcp")

@app.tool()
def my_tool(param: str) -> dict:
    """Tool description"""
    return {"result": "data"}

if __name__ == "__main__":
    app.run()  # Starts stdio server

特征：

• ✅ 自动JSON-RPC协议处理
• ✅ 工具发现(tools/list)
• ✅ 工具执行(tools/call)
• ✅ 输入/输出验证
• ✅ 错误处理

5.SQLite3

内置Python模块 -无需单独包装！

我们使用什么：

import sqlite3

# Read-only connection (security!)
uri = f"file:{path}?mode=ro"
conn = sqlite3.connect(uri, uri=True)

# Dict-like row access
conn.row_factory = sqlite3.Row

# Execute queries
cursor = conn.execute("SELECT * FROM movies WHERE year = ?", (1995,))
rows = cursor.fetchall()
for row in rows:
    print(row["title"])  # Dict-like access

安全功能：

• ✅ 只读模式(?mode=ro)
• ✅ 参数化查询(:param 语法）
• ✅ 连接上下文管理器（自动关闭）

🚀 设置和运行

先决条件

• Python 3.10+ （用3.13测试）
• 紫外线 包管理器（或pip）
• Google Gemini API密钥 （免费版可用）

步骤1：安装依赖项

使用紫外线（推荐）：

cd sqlite3-mcp
uv sync

使用pip：

pip install -e .

已安装的依赖项：

mcp>=1.20.0                 # Official MCP Python SDK
fastapi>=0.121.0            # Web framework
uvicorn>=0.38.0             # ASGI server
google-generativeai>=0.8.0  # Gemini SDK
python-dotenv>=1.0.0        # Environment variables

步骤2：获取Gemini API密钥

1. 首选https://aistudio.google.com/app/apikey
2. 单击“创建API密钥”
3. 复制密钥（以开头 AIza...)

步骤3：创建 .env 文件

创建一个名为的文件 .env 在项目根目录中：

# .env
LLM_API_KEY=AIzaSy...your-gemini-api-key-here...
PORT=8000

注： 这 .env 为了安全起见，文件被忽略了！

步骤4：运行服务器

uv run main.py

预期产量：

🚀 Starting SQLite3 MCP Chat Server (Official MCP Python SDK)...
📍 Server running at: http://localhost:8000
🔧 MCP Server: /path/to/src/sqlite3_mcp/server.py
💬 Open your browser to start chatting!

✅ Gemini API key configured
🤖 Model: gemini-2.0-flash-exp
🔗 Using Official MCP Python SDK (v1.20.0)

✅ MCP session initialized with 8 tools
   - ping: Quick health check
   - db_path: Return the absolute path of the SQLite database file
   - list_tables: List all user tables in the SQLite database
   - describe_table: Describe a table's columns
   - preview: Preview rows from a table
   - run_select: Run a read-only SELECT
   - run_exec: Execute a write statement
   - explain: Run EXPLAIN QUERY PLAN for a SELECT
✅ Gemini API configured (Pure MCP with Official SDK!)

INFO:     Started server process [12345]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)

步骤5：在浏览器中打开

导航到： http://localhost:8000

您应该看到一个漂亮的聊天界面：

• 欢迎信息
• 建议按钮
• 文本输入区
• 连接状态指示灯（绿点）

步骤6：尝试示例查询

数据库探索：

"What tables are in the database?"
"Describe the movies table"
"Show me 5 random movies"

简单查询：

"How many movies are there?"
"Find movies from 1995"
"List movies with 'love' in the title"

复杂查询（需要JOIN）：

"Who are the top 5 directors by number of movies?"
"Which directors made more than 10 movies?"
"Show me movies directed by Christopher Nolan"
"Find the most popular director"

聚合：

"How many movies per year?"
"What's the average movie rating?"
"Show me directors with highest average ratings"

步骤7：关闭服务器

方法1： 按 CTRL+C 在终端

方法2： 终止进程

lsof -ti:8000 | xargs kill -9

方法3： 手动查找并杀死

ps aux | grep "uv run main.py"
kill 

💡 示例问题库

初级查询

• “有什么桌子？”
• “给我看看电影表的结构”
• “数据库中有多少部电影？”
• “随机给我5个电影标题”

中间查询

• “查找1995年以来的所有电影”
• “列出收入超过1亿的电影”
• “给我看收视率最高的电影”
• “最受欢迎的电影是什么？”
• “查找标题中有‘Star’的电影”

高级查询（自动登录！）

• “谁导演的电影最多？”
• “按平均电影评级显示前10位导演”
• “1990年代有哪些导演拍过电影？”
• “寻找克里斯托弗·诺兰的电影作品”
• “谁是最多产的女导演？”

分析查询

• “每年电影的平均预算是多少？”
• “向我展示电影收视率随时间的变化趋势”
• “哪十年上映的电影最多？”
• “预算和收入之间有什么关系？”
• “寻找一贯制作高评价电影的导演”

创意查询

• “告诉我关于太空主题的电影”
• “有哪些隐藏的宝石（高评级，低人气）？”
• “寻找大片（高收入、高预算）”
• “给我看独立电影（低预算，高收视率）”

🔐 安全特性

1.只读数据库（默认）

# server.py
def _connect(readonly: bool = True):
    if readonly:
        uri = f"file:{DB_PATH.as_posix()}?mode=ro"  # Read-only!
        conn = sqlite3.connect(uri, uri=True)

优点：

• ✅ 不能意外修改数据
• ✅ 无法删除表格
• ✅ 无法创建新表
• ✅ 中仅包含SELECT查询 run_select()

注： run_exec() 该工具使用写模式，但需要明确的意图

2.SQL注入保护

# server.py
@app.tool()
def run_select(sql: str, params: Optional[Dict] = None, limit: int = 1000):
    # Parameterized queries - SQLite handles escaping
    cursor = conn.execute(sql, params or {})

安全使用：

# ✅ SAFE - parameterized
sql = "SELECT * FROM movies WHERE year = :year"
params = {"year": 1995}
run_select(sql, params)

不安全使用（不要这样做！）：

# ❌ DANGEROUS - SQL injection!
year = "1995 OR 1=1; DROP TABLE movies; --"
sql = f"SELECT * FROM movies WHERE year = {year}"
run_select(sql)  # Could drop table!

3.API密钥安全

# main.py
load_dotenv()  # Load from .env file
LLM_API_KEY = os.environ.get('LLM_API_KEY', None)

安全措施：

• ✅ API密钥存储在 .env 文件（gitignored）
• ✅ 从未接触过前端
• ✅ 仅使用服务器端
• ✅ 未登录输出

4.MCP协议隔离

# MCP server runs as subprocess
subprocess: python server.py
stdin/stdout communication only

优点：

• ✅ 服务器在单独的进程中运行
• ✅ 无法访问主进程内存
• ✅ 仅限于定义的工具
• ✅ 崩溃不会影响主服务器

5.输入验证

# Pydantic models
class MessageRequest(BaseModel):
    message: str
    history: List[Dict] = []

自动验证：

• ✅ 类型检查
• ✅ 必填字段
• ✅ 字段约束
• ✅ 防止格式错误的请求

6.速率限制（Gemini）

• 谷歌强制执行API费率限制
• 免费等级：约60个请求/分钟
• 考虑为生产环境实施服务器端速率限制

🐛 故障排除

“未配置LLM API密钥”

症状：

⚠️  Warning: LLM_API_KEY not found in .env

解决：

1. 检查 .env 文件存在于项目根目录中
2. 验证变量名是否为 LLM_API_KEY （区分大小写！）
3. 验证密钥周围没有引号： LLM_API_KEY=AIzaSy...
4. 更改后重新启动服务器 .env

测试：

cat .env  # Should show: LLM_API_KEY=AIza...

“MCP会话未初始化”

症状：

❌ Error initializing MCP: [error details]
ERROR:    Application startup failed. Exiting.

常见原因：

1. Python版本太旧 -需要Python 3.10+

   python --version  # Should be 3.10+

1. MCP服务器路径错误

   ls src/sqlite3_mcp/server.py  # Should exist

1. 数据库文件丢失

   ls movies.sqlite  # Should exist

1. 未安装依赖项

   uv sync  # or pip install -e .

“anyio.BrokenResourceError”

症状：

anyio.BrokenResourceError
ERROR:    Application startup failed.

原因： 异步上下文管理不当

解决方案： 这应该在当前代码中修复！我们使用 AsyncExitStack:

@asynccontextmanager
async def lifespan(app: FastAPI):
    mcp_exit_stack = AsyncExitStack()
    
    read, write = await mcp_exit_stack.enter_async_context(
        stdio_client(server_params)
    )
    mcp_session = await mcp_exit_stack.enter_async_context(
        ClientSession(read, write)
    )
    
    yield
    
    await mcp_exit_stack.aclose()  # Proper cleanup!

如果您看到此错误，请确保您使用的是最新代码。

“429资源枯竭”（双子座）

症状：

Error: 429 Resource exhausted. Please try again later.

原因： 达到Gemini API利率限制

解决：

1. 等待60秒 然后重试
2. 检查您的配额 在https://aistudio.google.com/
3. 升级到付费等级 对于更高的限制

免费等级限制：

• 每分钟60个请求
• 每天1500个请求

“值错误：无法将part.function_call转换为文本”

症状：

ValueError: Could not convert `part.function_call` to text.

原因： 试图得到 .text 当响应仍有函数调用时

解决方案： 这应该在当前代码中修复！我们检查所有部件：

while iteration 抽象**