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

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

文章目录

  • 下载Trae
  • uv 工具教程参考我的这篇文章
  • 创建 uv 项目
  • main.py
  • Cherry Studio 添加 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/
--directory
~/TraeProjects/demo
run
main.py
--db-path
~/TraeProjects/demo/test.db

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

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 表中的数据

在这里插入图片描述

http://www.xdnf.cn/news/149329.html

相关文章:

  • Jetpack Compose 基础组件学习2.1:Surface
  • stack __ queue(栈和队列)
  • 分布式事务 两阶段提交协议(2PC的原理、挑战)
  • 大模型微调 - 自注意力机制
  • 【统计学习】递归最小二乘算法与奇异值分解
  • #什么是爬虫?——从技术原理到现实应用的全面解析 VI
  • Vue回调函数中的this
  • 【CF】Day43——Codeforces Round 906 (Div. 2) E1
  • Libconfig 修改配置文件里的某个节点
  • Linux 系统用户管理与权限掌控:从基础到精通
  • 《深入理解计算机系统》阅读笔记之第三章 程序的机器级表示
  • Python判断语句-语法:if,if else,if elif else,嵌套,if else语句扁平式写法,案例
  • LatentSync - 字节联合北交大开源的端到端唇形同步框架-附整合包
  • Cannot read properties of null (reading ‘classList‘)
  • 人工智能的100个关键词系统学习计划
  • Trae 实测:AI 助力前端开发,替代工具还远吗?
  • mysql 导入很慢,如何解决
  • 猿人学题库13题—动态css字体加密 记录
  • JavaScript性能优化实战(5):数据结构与算法性能优化
  • Python爬取天猫畅销榜接口的详细教程
  • Python基础语法:字符串格式化(占位拼接,精度控制,format()函数,快速格式化,表达式格式化)
  • dstream
  • 《深入浅出ProtoBuf:从环境搭建到高效数据序列化》​
  • python基础-requests结合AI实现自动化数据抓取
  • 文档编辑:reStructuredText全面使用指南 — 第三部分 进阶特性
  • 第四章 安全审计
  • HMI与组态,自动化的“灵珠”和“魔丸”
  • 【FastJSON】的parse与parseObject
  • Huffman(哈夫曼)解/压缩算法实现
  • 【多目标进化算法】常见多目标进化算法一览