使用说明

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

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

🎯 项目概述

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

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

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

✨ 主要特点

特性	描述
🔗 官方MCP SDK	使用标准 `mcp` 用于协议兼容通信的Python包
🎯 Gemini本地函数调用	与Gemini的函数调用API直接集成- 没有框架像LangChain
🔍 动态模式发现	AI使用MCP工具实时探索数据库结构
🔄 基于会话的历史记录	服务器重启时自动清除聊天记录
💬 现代用户界面	漂亮、反应灵敏的聊天界面，配有打字指示器
🔐 缺省巩固安全	只读数据库连接、参数化查询、API密钥保护
⚡ 生产准备就绪	正确的异步上下文管理、错误处理、类型安全

🎬 交互示例

简单模式发现

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()

幕后发生了什么：

stdio_client(server_params) 生成一个子流程： python server.py
子进程运行MCP服务器，监听stdin/stdout
ClientSession(read, write) 建立双向通信

initialize() 发送以下JSON-RPC消息：

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

list_tools() 发送：

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

服务器返回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."""
)

为什么这个系统会提示？

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

步骤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 < max_iterations:  # max 10 iterations
    # Check ALL parts for function calls (robust checking)
    has_function_call = False
    function_call = None
    
    for part in response.candidates[0].content.parts:
        if part.function_call:
            has_function_call = True
            function_call = part.function_call
            break
    
    if not has_function_call:
        break  # No more function calls - we're done!
    
    iteration += 1

为什么要检查所有零件？

双子座的反应可以有多个部分
函数调用可能不在 parts[0]
仅检查 parts[0] 原因 ValueError: Could not convert part.function_call to text

步骤5.2：提取函数详细信息

# main.py (lines 297-301)
tool_name = function_call.name  # "describe_table"
tool_args = dict(function_call.args)  # {"table": "movies"}

print(f"🔧 Iteration {iteration}: Calling tool {tool_name} with args {tool_args}")

步骤5.3：通过官方SDK调用MCP工具

# main.py (lines 144-173)
async def call_mcp_tool(tool_name: str, arguments: Dict[str, Any]) -> 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=<your-gemini-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？

Aspect	官方SDK	手动JSON-RPC
正确性	✅ 符合规范	❌ 容易出错
类型安全	✅ Pydantic模型	❌ 原始字典
错误处理	✅ 内置	❌ 手册
维护	✅ 已更新规格	❌ 你维护它
异步	✅ 正确的上下文管理	❌ 资源泄漏

了解更多：

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密钥

首选https://aistudio.google.com/app/apikey
单击“创建API密钥”
复制密钥（以开头 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 <PID>

💡 示例问题库

初级查询

“有什么桌子？”
“给我看看电影表的结构”
“数据库中有多少部电影？”
“随机给我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

解决：

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

测试：

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

“MCP会话未初始化”

症状：

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

常见原因：

Python版本太旧 -需要Python 3.10+
```
python --version  # Should be 3.10+
```

MCP服务器路径错误

ls src/sqlite3_mcp/server.py  # Should exist

数据库文件丢失
```
ls movies.sqlite  # Should exist
```
未安装依赖项
```
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利率限制

解决：

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

免费等级限制：

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

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

症状：

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

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

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

while iteration < max_iterations:
    has_function_call = False
    
    # Check ALL parts (not just parts[0])
    for part in response.candidates[0].content.parts:
        if part.function_call:
            has_function_call = True
            break
    
    if not has_function_call:
        break  # Safe to call response.text now!

response_text = response.text  # Only after loop exits

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

工具未被调用

症状：

人工智能以对话方式回应，但不使用工具
没有🔧 迭代X：调用工具。..“在日志中

解决：

检查Gemini模型是否支持函数调用

model_name='gemini-2.0-flash'  # ✅ Supports function calling
# Not 'gemini-pro' (older model)

检查系统提示是否清晰

system_instruction="...Be proactive - use the tools immediately..."

尝试更明确的问题

❌ "Tell me about movies"  # Too vague
✅ "List the tables in the database"  # Clear intent

检查服务器日志 MCP通信错误

AI显示SQL代码

症状：

AI响应包括SQL查询
用户看到： SELECT * FROM movies...

原因： 系统提示不够强或AI忽略指令

解决：

更新系统提示 （已在当前代码中完成）：

system_instruction="...Never show SQL code to the user - only show the results..."

重新表述问题：

❌ "Write a query to find..."
✅ "Show me the..."

Check AI实际上正在调用工具 （检查日志 🔧 Iteration)

不正确的SQL查询

症状：

Error: no such column: director
Error: no such table: movie

原因： AI不理解模式或关系

解决：

系统提示包括关系提示 （已完成）：

system_instruction="...The movies table has a director_id foreign key 
                      that links to the directors table's id..."

Check AI正在调用模式发现工具：
- 应该打电话 list_tables()
- 应该打电话 describe_table() 之前 run_select()
验证数据库架构是否与提示匹配：
```
sqlite3 movies.sqlite ".schema"
```

重新启动时聊天记录未清除

症状：

服务器重新启动，但旧聊天记录仍然存在
没有🔄 检测到服务器重新启动”消息

解决方案： 这应该在当前代码中修复！

它是如何工作的：

// Frontend checks session ID on every interaction
async function checkServerRestart() {
    const data = await fetch('/api/status').json();
    const serverSessionId = data.session_id;
    const storedSessionId = localStorage.getItem('serverSessionId');
    
    if (storedSessionId && storedSessionId !== serverSessionId) {
        // Clear history!
        clearChatHistory();
        showRestartNotification();
    }
    
    localStorage.setItem('serverSessionId', serverSessionId);
}

手动修复：

打开浏览器控制台（F12）
运行： localStorage.clear()
刷新页面

端口8000已在使用中

症状：

ERROR:    [Errno 48] Address already in use

解决：

终止端口8000上的进程：
```
lsof -ti:8000 | xargs kill -9
```
更改端口 .env:
```
PORT=8001
```

手动查找并杀死：

lsof -i:8000  # Find process ID
kill <PID>

📚 了解更多

官方文件

MCP规范： https://spec.modelcontextprotocol.io/
MCP Python SDK： https://github.com/modelcontextprotocol/python-sdk
MCP概述： https://modelcontextprotocol.io/
Gemini函数调用： https://ai.google.dev/gemini-api/docs/function-calling
FastAPI文档： https://fastapi.tiangolo.com/

教程和指南

MCP快速入门： https://modelcontextprotocol.io/quickstart
构建MCP服务器： https://modelcontextprotocol.io/docs/concepts/servers
构建MCP客户端： https://modelcontextprotocol.io/docs/concepts/clients
FastAPI生命周期事件： https://fastapi.tiangolo.com/advanced/events/

🎓 关键设计决策

为什么没有LangChain？

LangChain非常适合：

快速原型制作
具有多个步骤的复杂代理工作流
思维链推理
许多开箱即用的集成

我们选择不使用它，因为：

更简单的架构 -更少的抽象层，更容易理解
直接控制 -完全控制函数调用循环
学习 -更好地理解MCP协议和Gemini API
演出 -无框架开销
依赖项 -需要管理的包更少（4个vs 10+）

我们的方法：

# ✅ Our way: Direct Gemini + MCP SDK
model = genai.GenerativeModel(tools=mcp_tools)
response = chat.send_message(message)
if response.function_call:
    result = await mcp_session.call_tool(name, args)
    response = chat.send_message(function_response=result)

郎链方式：

# ❌ LangChain way: More abstraction
from langchain.agents import create_agent
from langchain_mcp_adapters import MultiServerMCPClient

mcp_client = MultiServerMCPClient(...)
tools = await mcp_client.get_tools()
agent = create_agent(model, tools)
result = await agent.ainvoke(message)

何时使用LangChain：

构建具有记忆、规划和反射的复杂代理
需要许多集成（矢量存储、文档加载器等）
快速原型制作，无需关心内部

何时使用我们的方法：

想了解它是如何工作的
需要最大程度的控制
构建具有特定要求的生产系统
学习MCP协议

为什么选择官方MCP SDK？

我们最初实现了手动JSON-RPC：

# ❌ Manual JSON-RPC (our first attempt)
request = {"jsonrpc": "2.0", "method": "tools/call", ...}
mcp_process.stdin.write(json.dumps(request) + "\n")
response = json.loads(mcp_process.stdout.readline())

然后转到官方SDK：

# ✅ Official SDK (current implementation)
result = await mcp_session.call_tool(name, args)

我们为什么切换：

Aspect	官方SDK	手动JSON-RPC
正确性	✅ 符合规范	❌ 容易出错
类型安全	✅ Pydantic模型	❌ 原始字典
错误处理	✅ 内置	❌ 手动尝试/例外
维护	✅ 已更新规格	❌ 你维护它
异步	✅ 异步退出堆栈	❌ 资源泄漏
文档	✅ 有据可查	❌ DIY文档

何时使用手动JSON-RPC：

学习MCP协议的工作原理
调试协议问题
实施定制运输

何时使用官方SDK：

生产系统（始终如此！）
想要可靠性和正确性
需要良好的错误消息
想要类型安全

为什么选择AsyncExitStack？

我们面临的问题：

# ❌ Manual context managers (caused BrokenResourceError)
read, write = await stdio_client(server_params).__aenter__()
mcp_session = await ClientSession(read, write).__aenter__()
# No proper cleanup! Resources leak!

解决方案：

# ✅ AsyncExitStack (proper resource management)
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  # Server runs

await mcp_exit_stack.aclose()  # Guaranteed cleanup!

优点：

✅ 适当的资源清理（始终运行）
✅ 不 BrokenResourceError
✅ 优雅地处理异常
✅ 标准Python模式

为什么要重置基于会话的历史记录？

问题：

用户与AI讨论数据库
服务器重新启动（代码更改、崩溃等）
旧聊天记录仍在本地存储中
用户继续对话，AI感到困惑（新的MCP会话！）

解决方案：

# Backend: Generate unique session ID on startup
SERVER_SESSION_ID = str(int(time.time()))  # Unix timestamp

@app.get("/api/status")
async def status():
    return {"session_id": SERVER_SESSION_ID, ...}

// Frontend: Check session ID on every interaction
const serverSessionId = await fetch('/api/status').session_id;
const storedSessionId = localStorage.getItem('serverSessionId');

if (storedSessionId !== serverSessionId) {
    clearChatHistory();  // Server restarted!
}

优点：

✅ 重新启动后没有过时的数据
✅ 用户知道服务器已重新启动（视觉通知）
✅ 重新开始，没有混乱
✅ 自动工作（无需用户操作）

📊 性能注意事项

典型响应时间

问：“电影数量排名前五的导演是谁？”

故障：

前端→ 后端： ~50ms（网络）
工具调用：
- list_tables()：~50ms（SQLite查询）
- describe_table("movies")：~100ms（PRAGMA查询）
- describe_table("directors")：~100ms
- run_select(sql)：~150ms（JOIN查询）
Gemini处理： 每次通话约500-1000ms
- 呼叫1：决定使用 list_tables()
- 调用2：分析、使用 describe_table()
- 呼叫3：分析、使用 describe_table()
- 调用4：生成SQL，使用 run_select()
- 调用5：格式化结果（无工具调用）
后端→ 前端： ~50ms（网络）

总计： 2-4秒用于复杂的多工具查询 简单查询（无工具）： <1秒

优化机会

1.模式缓存

# Current: Call describe_table() every time
schema = await mcp_session.call_tool("describe_table", {"table": "movies"})

# Optimization: Cache in memory
schema_cache = {}
if table not in schema_cache:
    schema_cache[table] = await mcp_session.call_tool(...)
schema = schema_cache[table]

权衡：

✅ 更快的响应（无重复的PRAGMA查询）
❌ 如果架构更改，则数据将过时

2.连接池

# Current: New connection per tool call
with _connect(readonly=True) as conn:
    rows = conn.execute(sql)

# Optimization: Reuse connections
conn_pool = []  # Pool of connections
conn = conn_pool.pop() or _connect()
rows = conn.execute(sql)
conn_pool.append(conn)

权衡：

✅ 减少连接开销
❌ 更复杂的代码
❌ 需要连接健康检查

3.快速工程

# Current: Generic system prompt
system_instruction = "Use tools to discover schema..."

# Optimization: Provide common patterns
system_instruction = """
Common queries:
- Top directors: SELECT d.name, COUNT(*) FROM movies m JOIN directors d ...
- Find director: SELECT m.title FROM movies m JOIN directors d WHERE d.name LIKE :name
...
"""

权衡：

✅ AI减少了工具调用
❌ 动态性较低，可能不适用于所有查询
❌ 更长的提示=更高的成本

4.流媒体响应

# Current: Wait for full response
response = chat.send_message(message)
return response.text

# Optimization: Stream as AI generates
for chunk in chat.send_message_stream(message):
    yield chunk.text

权衡：

✅ 更好的用户体验（用户逐渐看到响应）
✅ 感觉更快
❌ 更复杂的前端代码
❌ 函数调用时不能轻易中断

5.在Prompt中预加载架构

# Optimization: Include schema in system prompt
system_instruction = f"""
Database schema:
- movies table: {schema_movies}
- directors table: {schema_directors}
...
"""

权衡：

✅ 不需要 describe_table() 电话
❌ 更长的提示=更高的成本
❌ 反对MCP哲学（动态发现）
❌ 如果架构更改，则处于停滞状态

我们的选择： 我们优先考虑 正确性与MCP哲学 超性能。当前的实现对于大多数用例来说足够快（对于复杂的查询，需要2-4s）。

🤝 贡献

欢迎投稿！以下是一些想法：

简单任务

[]在README中添加更多示例查询
[]改进错误消息
[]在UI中添加加载状态
[]添加键盘快捷键

中等任务

[]实现模式缓存
[]添加速率限制
[]为工具添加单元测试
[]支持其他数据库（Postgres、MySQL）

艰巨的任务

[]添加流媒体响应
[]添加身份验证
[]部署到生产环境（Docker、云）
[]添加聊天导出/导入
[]多用户支持

如何做出贡献：

分叉存储库
创建要素分支： git checkout -b feature/amazing-feature
进行更改
彻底测试
承诺： git commit -m 'Add amazing feature'
推： git push origin feature/amazing-feature
打开拉取请求

📄 许可证

MIT许可证-请随意使用、修改和分发！

🙏 致谢

内置❤️ 使用：

模型上下文协议（MCP） -Anthropic的人工智能工具通信开放标准
MCP官方Python SDK -模型上下文协议团队
谷歌双子座2.0 Flash -谷歌最新的快速LLM函数调用
快速API -Sebastián Ramírez（@tiangolo）对令人惊叹的web框架表示感谢
FastMCP -用于构建服务器的官方MCP SDK的一部分
SQLite -D.Richard Hipp为令人难以置信的嵌入式数据库

特别感谢开源社区使这一切成为可能！

📞 支持和问题

问题： 在GitHub上打开一个问题
问题： 请先阅读故障排除部分
文档： https://modelcontextprotocol.io/

🎯 快速命令参考

# Install dependencies
uv sync

# Run server
uv run main.py

# Kill server
lsof -ti:8000 | xargs kill -9

# Check if server is running
curl http://localhost:8000/api/status

# Test chat endpoint
curl -X POST http://localhost:8000/api/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "What tables are in the database?", "history": []}'

# Run MCP server standalone (for testing)
python src/sqlite3_mcp/server.py

# Check database schema
sqlite3 movies.sqlite ".schema"

# Check database tables
sqlite3 movies.sqlite "SELECT name FROM sqlite_master WHERE type='table';"

# Count movies
sqlite3 movies.sqlite "SELECT COUNT(*) FROM movies;"

# List dependencies
uv pip list

# Check Python version
python --version

最后更新时间： 2025年1月
MCP SDK版本： 1.20.0+
双子座模型： 双子座2.0闪光
FastAPI版本： 0.121.0+

采用纯MCP+Gemini原生函数调用构建

✨ 无LangChain•无框架•直接协议•生产就绪✨

理解>抽象

使用说明

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

🎯 项目概述

✨ 主要特点

🎬 交互示例

简单模式发现

使用JOIN的复杂查询

🏗️ 建筑

🔧 完整的生命周期说明

第一阶段：服务器启动

步骤1.1:FastAPI初始化

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

步骤1.3：服务器就绪

第二阶段：用户交互

步骤2.1：前端初始化

步骤2.2：会话重启检测

步骤2.3：用户发送消息

阶段3：后端处理

步骤3.1：接收请求

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

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

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

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

第四阶段：Gemini处理

阶段5：函数调用循环

步骤5.1：检测函数调用

步骤5.2：提取函数详细信息

步骤5.3：通过官方SDK调用MCP工具

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

步骤5.5：将结果发送回Gemini

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

步骤5.7：退出循环

步骤5.8：提取最终响应

第六阶段：返回前端

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

📁 项目结构

🔑 技术栈深潜

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

2.谷歌双子座2.0 Flash

3.FastAPI

4.FastMCP（MCP服务器SDK）

5.SQLite3

🚀 设置和运行

先决条件

步骤1：安装依赖项

步骤2：获取Gemini API密钥

步骤3：创建 .env 文件

步骤4：运行服务器

步骤5：在浏览器中打开

步骤6：尝试示例查询

步骤7：关闭服务器

💡 示例问题库

初级查询

中间查询

高级查询（自动登录！）

分析查询

创意查询

🔐 安全特性

1.只读数据库（默认）

2.SQL注入保护

3.API密钥安全

4.MCP协议隔离

5.输入验证

6.速率限制（Gemini）

🐛 故障排除

“未配置LLM API密钥”

“MCP会话未初始化”

“anyio.BrokenResourceError”

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

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

工具未被调用

AI显示SQL代码

不正确的SQL查询

重新启动时聊天记录未清除

端口8000已在使用中

📚 了解更多

官方文件

教程和指南

相关项目

🎓 关键设计决策

步骤3：创建 `.env` 文件