当前位置: 首页 > news >正文

使用 Python 项目管理工具 uv 快速创建 MCP 服务(Cherry Studio、Trae 添加 MCP 服务)

文章目录

  • 下载Trae
  • uv 工具教程参考我的这篇文章
  • 创建 uv 项目
  • main.py
  • Cherry Studio 添加 MCP 服务
    • DeepSeek API
    • 配置 DeepSeek API
    • 调用 MCP 服务
  • Trae 添加 MCP 服务
    • 添加 MCP
    • 创建智能体
  • 使用智能体
    • 调用 MCP 创建 demo 表
    • 查询 demo 表结构信息
    • demo 表插入 2 条测试数据
    • 查询 demo 表中的数据

下载Trae

  • https://www.trae.com.cn/
    在这里插入图片描述

uv 工具教程参考我的这篇文章

  • 让 Python 项目管理变简单(uv 工具快速上手指南)

创建 uv 项目

uv init demo
cd demo
  • 添加依赖项
uv add 'mcp[cli]'

在这里插入图片描述

main.py

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

Cherry Studio 添加 MCP 服务

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

  • MCP 参数

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

在这里插入图片描述
在这里插入图片描述

DeepSeek API

  • 申请 API_KEY:https://platform.deepseek.com/usage

在这里插入图片描述

配置 DeepSeek API

在这里插入图片描述

调用 MCP 服务

在这里插入图片描述
在这里插入图片描述

Trae 添加 MCP 服务

  • 下载Trae:https://www.trae.com.cn/

添加 MCP

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

在这里插入图片描述
在这里插入图片描述

创建智能体

在这里插入图片描述

使用智能体

调用 MCP 创建 demo 表

在这里插入图片描述

查询 demo 表结构信息

在这里插入图片描述

demo 表插入 2 条测试数据

在这里插入图片描述

查询 demo 表中的数据

在这里插入图片描述

相关文章:

  • 通道降维方式
  • 一款好的私有云产品推荐——优刻得私有云(UCloudStack Pro)产品白皮书
  • 单机无穷大系统暂态稳定性仿真Matlab模型
  • 数据库-子查询、关联查询 和 TCL 语言
  • 智慧医疗领域TMI期刊2025年3月研究热点解析
  • 嵌入式:Linux系统应用程序(APP)启动参数及其规则详解
  • 【网络入侵检测】基于源码分析Suricata的PCAP模式
  • 计算器(WEB)
  • 流动式起重机Q2证考试有哪些科目?
  • C++与Python编写二进制转十进制
  • 机器人行业研究系列报告
  • 方案精读:77页2024 集团企业IT技术架构规划方案【附全文阅读】
  • IntelliJ IDEA 中配置 Spring MVC 环境的详细步骤
  • Video-LLaVA
  • DeepSeek系列(8):个人效能提升案例
  • 《使用 Cesium 加载静态热力图显示的实现步骤》
  • 前端如何获取文件的 Hash 值?多种方式详解、对比与实践指南
  • 实战指南:搭建智慧变电站管理平台全流程解析(二)
  • HFSS5(李明洋)——设置激励(波端口激励)
  • 异步请求池控制同一时间并发
  • 时代邻里:拟收购成都合达联行科技剩余20%股权
  • 又双叒叕出差太空了!神二十成功出发,神十九乘组扫榻以待
  • 阻燃材料点火就着引发一场火灾,河北一企业的产品被指不达标且涉嫌欺诈
  • 广东江门公布“小客车坠海致3死”事故评估报告,司机被判三年缓五年
  • 书信里的宋人|忠奸难辨的夏竦
  • 男粉丝咬伤女主播嘴后写的条子引争议:赔偿“十万元”还是“十5元”?