04_MCP

MCP 简介**MCPModel Context Protocol模型上下文协议**为 AI 应用提供外部数据源和工具访问能力的程序可以把 MCP 理解为AI 应用的 USB-C 接口——提供一种标准化方式让 AI 连接外部系统。三层角色MCP的架构其实很简单就三个角色MCP Client使用工具的AI大脑就是你用的AI工具比如Cursor、Claude Desktop。它负责理解你的需求决定要不要用工具。MCP Server提供工具的服务员这是你配置的服务专门给AI提供某个能力。比如数据库MCP Server让AI能查数据库API MCP Server让AI能调接口。Tools具体的能力每个MCP Server暴露的具体功能。比如数据库Server可能有”查询表结构”、“执行SELECT查询”、“统计行数”这些工具。类比一下AI是个工人ClientMCP Server是工具箱里面有扳手、锤子Tools。你说要钉钉子AI就知道从工具箱里拿锤子出来用。三大核心能力1.工具 (Tools)AI 可以调用执行的函数用于产生操作或获取实时数据。2. 资源 (Resources)为 AI 提供上下文信息的数据源是只读的。3. 提示词 (Prompts)可重用的交互模板帮助构建与 LLM 的标准化交互。典型场景系统提示词、少样本示例、特定领域的交互模板配置MCP Server全局配置所有项目都能用Windows:C:\Users\你的用户名\.cursor\mcp.jsonMac/Linux:~/.cursor/mcp.json项目级配置只在当前项目生效项目根目录下的.cursor/mcp.json创建.cursor/mcp.json文件写入以下内容{mcpServers:{sqlite:{command:npx,args:[-y,modelcontextprotocol/server-sqlite,--db-path,D:/path/to/your/test.db]}}}关键点解释mcpServers: 这是固定的字段名别改sqlite: 这是你给这个Server起的名字随便取AI会看到这个名字command: npx: 用npx直接运行MCP Server不用手动安装args: 传给命令的参数-y: 自动确认安装modelcontextprotocol/server-sqlite: 这是官方的SQLite MCP Server包--db-path: 数据库文件路径必须是绝对路径Windows用户注意路径要用正斜杠/或双反斜杠\\不能用单反斜杠✅D:/projects/test.db✅D:\\\\projects\\\\test.db❌D:\projects\test.db这会报错自定义MCP Server用python FastMCP开发一个本地MCP服务frompathlibimportPathimportsqlite3importosfromtypingimportList,Dict,Any,OptionalfromfastmcpimportFastMCP# Initialize FastMCP servermcpFastMCP(SQLite Explorer,log_levelCRITICAL)# Path to Messages database - must be provided via SQLITE_DB_PATH environment variableifSQLITE_DB_PATHnotinos.environ:raiseValueError(SQLITE_DB_PATH environment variable must be set)DB_PATHPath(os.environ[SQLITE_DB_PATH])classSQLiteConnection:def__init__(self,db_path:Path):self.db_pathdb_path self.connNonedef__enter__(self):self.connsqlite3.connect(str(self.db_path))self.conn.row_factorysqlite3.Rowreturnself.conndef__exit__(self,exc_type,exc_val,exc_tb):ifself.conn:self.conn.close()mcp.tool()defread_query(query:str,params:Optional[List[Any]]None,fetch_all:boolTrue,row_limit:int1000)-List[Dict[str,Any]]:Execute a query on the Messages database. Args: query: SELECT SQL query to execute params: Optional list of parameters for the query fetch_all: If True, fetches all results. If False, fetches one row. row_limit: Maximum number of rows to return (default 1000) Returns: List of dictionaries containing the query results ifnotDB_PATH.exists():raiseFileNotFoundError(fMessages database not found at:{DB_PATH})# Clean and validate the queryqueryquery.strip()# Remove trailing semicolon if presentifquery.endswith(;):queryquery[:-1].strip()# Check for multiple statements by looking for semicolons not inside quotesdefcontains_multiple_statements(sql:str)-bool:in_single_quoteFalsein_double_quoteFalseforcharinsql:ifcharandnotin_double_quote:in_single_quotenotin_single_quoteelifcharandnotin_single_quote:in_double_quotenotin_double_quoteelifchar;andnotin_single_quoteandnotin_double_quote:returnTruereturnFalseifcontains_multiple_statements(query):raiseValueError(Multiple SQL statements are not allowed)# Validate query type (allowing common CTEs)query_lowerquery.lower()ifnotany(query_lower.startswith(prefix)forprefixin(select,with)):raiseValueError(Only SELECT queries (including WITH clauses) are allowed for safety)paramsparamsor[]withSQLiteConnection(DB_PATH)asconn:cursorconn.cursor()try:# Only add LIMIT if query doesnt already have oneiflimitnotinquery_lower:queryf{query}LIMIT{row_limit}cursor.execute(query,params)iffetch_all:resultscursor.fetchall()else:results[cursor.fetchone()]return[dict(row)forrowinresultsifrowisnotNone]exceptsqlite3.Errorase:raiseValueError(fSQLite error:{str(e)})mcp.tool()deflist_tables()-List[str]:List all tables in the Messages database. Returns: List of table names in the database ifnotDB_PATH.exists():raiseFileNotFoundError(fMessages database not found at:{DB_PATH})withSQLiteConnection(DB_PATH)asconn:cursorconn.cursor()try:cursor.execute( SELECT name FROM sqlite_master WHERE typetable ORDER BY name )return[row[name]forrowincursor.fetchall()]exceptsqlite3.Errorase:raiseValueError(fSQLite error:{str(e)})mcp.tool()defdescribe_table(table_name:str)-List[Dict[str,str]]:Get detailed information about a tables schema. Args: table_name: Name of the table to describe Returns: List of dictionaries containing column information: - name: Column name - type: Column data type - notnull: Whether the column can contain NULL values - dflt_value: Default value for the column - pk: Whether the column is part of the primary key ifnotDB_PATH.exists():raiseFileNotFoundError(fMessages database not found at:{DB_PATH})withSQLiteConnection(DB_PATH)asconn:cursorconn.cursor()try:# Verify table existscursor.execute( SELECT name FROM sqlite_master WHERE typetable AND name? ,[table_name])ifnotcursor.fetchone():raiseValueError(fTable {table_name} does not exist)# Get table schemacursor.execute(fPRAGMA table_info({table_name}))columnscursor.fetchall()return[dict(row)forrowincolumns]exceptsqlite3.Errorase:raiseValueError(fSQLite error:{str(e)})mcp.tool()definsert_data(table_name:str,data:Dict[str,Any])-str:Insert data into a specified table. Args: table_name: Name of the table to insert data into. data: A dictionary where keys are column names and values are the data to insert. Returns: A message indicating success or failure. ifnotDB_PATH.exists():raiseFileNotFoundError(fMessages database not found at:{DB_PATH})withSQLiteConnection(DB_PATH)asconn:cursorconn.cursor()try:columns, .join(data.keys())placeholders, .join([?for_indata])sqlfINSERT INTO{table_name}({columns}) VALUES ({placeholders})cursor.execute(sql,list(data.values()))conn.commit()# Commit the transactionreturnfData inserted successfully into {table_name}.exceptsqlite3.Errorase:raiseValueError(fSQLite error:{str(e)})mcp.tool()defupdate_data(table_name:str,updates:Dict[str,Any],condition:str)-str:Update data in a specified table. Args: table_name: Name of the table to update. updates: A dictionary where keys are column names and values are the new data. condition: The condition for which rows to update (e.g., id 1). Returns: A message indicating success or failure. ifnotDB_PATH.exists():raiseFileNotFoundError(fMessages database not found at:{DB_PATH})withSQLiteConnection(DB_PATH)asconn:cursorconn.cursor()try:updates_string, .join([f{key} ?forkeyinupdates.keys()])sqlfUPDATE{table_name}SET{updates_string}WHERE{condition}cursor.execute(sql,list(updates.values()))conn.commit()# Commit the transactionreturnfData updated successfully in {table_name} where{condition}.exceptsqlite3.Errorase:raiseValueError(fSQLite error:{str(e)})mcp.tool()defdelete_data(table_name:str,condition:str)-str:Delete data from a specified table. Args: table_name: Name of the table to delete data from. condition: The condition for which rows to delete (e.g., id 1). Returns: A message indicating success or failure. ifnotDB_PATH.exists():raiseFileNotFoundError(fMessages database not found at:{DB_PATH})withSQLiteConnection(DB_PATH)asconn:cursorconn.cursor()try:sqlfDELETE FROM{table_name}WHERE{condition}cursor.execute(sql)conn.commit()# Commit the transactionreturnfData deleted successfully from {table_name} where{condition}.exceptsqlite3.Errorase:raiseValueError(fSQLite error:{str(e)})在cursor里添加mcp server{mcpServers:{sqlite-explorer:{command:uv,args:[run,--with,fastmcp,--with,uvicorn,fastmcp,run,E:/github/sqlite-explorer-fastmcp-mcp-server/sqlite_explorer.py],env:{SQLITE_DB_PATH:c:/users/user/my.db}}}}