04_MCP
2026/7/5 8:44:28 网站建设 项目流程

MCP 简介

**MCP(Model 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是个工人(Client),MCP Server是工具箱,里面有扳手、锤子(Tools)。你说要钉钉子,AI就知道从工具箱里拿锤子出来用。

三大核心能力

1.工具 (Tools)

AI 可以调用执行的函数,用于产生操作或获取实时数据。

2. 资源 (Resources)

为 AI 提供上下文信息的数据源,是只读的。

3. 提示词 (Prompts)

可重用的交互模板,帮助构建与 LLM 的标准化交互。

典型场景:系统提示词、少样本示例、特定领域的交互模板

配置MCP Server

全局配置(所有项目都能用):

  • Windows:C:\Users\你的用户名\.cursor\mcp.json
  • Mac/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"]}}}

关键点解释:

  1. mcpServers: 这是固定的字段名,别改
  2. "sqlite": 这是你给这个Server起的名字,随便取,AI会看到这个名字
  3. command: "npx": 用npx直接运行MCP Server,不用手动安装
  4. 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 servermcp=FastMCP("SQLite Explorer",log_level="CRITICAL")# Path to Messages database - must be provided via SQLITE_DB_PATH environment variableif'SQLITE_DB_PATH'notinos.environ:raiseValueError("SQLITE_DB_PATH environment variable must be set")DB_PATH=Path(os.environ['SQLITE_DB_PATH'])classSQLiteConnection:def__init__(self,db_path:Path):self.db_path=db_path self.conn=Nonedef__enter__(self):self.conn=sqlite3.connect(str(self.db_path))self.conn.row_factory=sqlite3.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:bool=True,row_limit:int=1000)->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(f"Messages database not found at:{DB_PATH}")# Clean and validate the queryquery=query.strip()# Remove trailing semicolon if presentifquery.endswith(';'):query=query[:-1].strip()# Check for multiple statements by looking for semicolons not inside quotesdefcontains_multiple_statements(sql:str)->bool:in_single_quote=Falsein_double_quote=Falseforcharinsql:ifchar=="'"andnotin_double_quote:in_single_quote=notin_single_quoteelifchar=='"'andnotin_single_quote:in_double_quote=notin_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_lower=query.lower()ifnotany(query_lower.startswith(prefix)forprefixin('select','with')):raiseValueError("Only SELECT queries (including WITH clauses) are allowed for safety")params=paramsor[]withSQLiteConnection(DB_PATH)asconn:cursor=conn.cursor()try:# Only add LIMIT if query doesn't already have oneif'limit'notinquery_lower:query=f"{query}LIMIT{row_limit}"cursor.execute(query,params)iffetch_all:results=cursor.fetchall()else:results=[cursor.fetchone()]return[dict(row)forrowinresultsifrowisnotNone]exceptsqlite3.Errorase:raiseValueError(f"SQLite 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(f"Messages database not found at:{DB_PATH}")withSQLiteConnection(DB_PATH)asconn:cursor=conn.cursor()try:cursor.execute(""" SELECT name FROM sqlite_master WHERE type='table' ORDER BY name """)return[row['name']forrowincursor.fetchall()]exceptsqlite3.Errorase:raiseValueError(f"SQLite error:{str(e)}")@mcp.tool()defdescribe_table(table_name:str)->List[Dict[str,str]]:"""Get detailed information about a table's 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(f"Messages database not found at:{DB_PATH}")withSQLiteConnection(DB_PATH)asconn:cursor=conn.cursor()try:# Verify table existscursor.execute(""" SELECT name FROM sqlite_master WHERE type='table' AND name=? """,[table_name])ifnotcursor.fetchone():raiseValueError(f"Table '{table_name}' does not exist")# Get table schemacursor.execute(f"PRAGMA table_info({table_name})")columns=cursor.fetchall()return[dict(row)forrowincolumns]exceptsqlite3.Errorase:raiseValueError(f"SQLite 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(f"Messages database not found at:{DB_PATH}")withSQLiteConnection(DB_PATH)asconn:cursor=conn.cursor()try:columns=', '.join(data.keys())placeholders=', '.join(['?'for_indata])sql=f"INSERT INTO{table_name}({columns}) VALUES ({placeholders})"cursor.execute(sql,list(data.values()))conn.commit()# Commit the transactionreturnf"Data inserted successfully into '{table_name}'."exceptsqlite3.Errorase:raiseValueError(f"SQLite 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(f"Messages database not found at:{DB_PATH}")withSQLiteConnection(DB_PATH)asconn:cursor=conn.cursor()try:updates_string=', '.join([f"{key}= ?"forkeyinupdates.keys()])sql=f"UPDATE{table_name}SET{updates_string}WHERE{condition}"cursor.execute(sql,list(updates.values()))conn.commit()# Commit the transactionreturnf"Data updated successfully in '{table_name}' where{condition}."exceptsqlite3.Errorase:raiseValueError(f"SQLite 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(f"Messages database not found at:{DB_PATH}")withSQLiteConnection(DB_PATH)asconn:cursor=conn.cursor()try:sql=f"DELETE FROM{table_name}WHERE{condition}"cursor.execute(sql)conn.commit()# Commit the transactionreturnf"Data deleted successfully from '{table_name}' where{condition}."exceptsqlite3.Errorase:raiseValueError(f"SQLite 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"}}}}

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询