services:
  mysql:
    image: mysql:5.7
    container_name: mysql5.7
    ports:
      - "3306:3306"
    environment:
      - MYSQL_ROOT_PASSWORD=root
      - MYSQL_ALLOW_EMPTY_PASSWORD=yes
      - TZ=Asia/Shanghai
    volumes:
      - ./volumes:/var/lib/mysql
    command: --character-set-server=utf8mb4

create database testdb;

use testdb;

CREATE TABLE `finance` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `user_id` varchar(36) NOT NULL DEFAULT '' COMMENT '用戶ID',
  `date` datetime NOT NULL COMMENT '金額發(fā)生日期',
  `amount` decimal(10,2) NOT NULL DEFAULT '0.00' COMMENT '收入支出金額(收入記為正數(shù)，支出記為負(fù)數(shù))',
  `category` varchar(32) NOT NULL DEFAULT '' COMMENT '收支類別',
  `remark` varchar(100) NOT NULL DEFAULT '' COMMENT '收支具體類目',
  PRIMARY KEY (`id`),
  KEY `idx_user_date` (`user_id`,`date`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='日常收支';

powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"

# Create a new directory for our project
uv init finance
cd finance

# Create virtual environment and activate it
uv venv
.venv\Scripts\activate

# Install dependencies
uv add mcp[cli] mysql-connector-python python-dotenv

DB_HOST=localhost
DB_USER=root
DB_PASSWD=root
DB_NAME=testdb

import mysql.connector
import json
import os
from contextlib import contextmanager
from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP

# 加載環(huán)境變量
load_dotenv()


class DbManager:
    def __init__(self):
        self.connection_pool = mysql.connector.pooling.MySQLConnectionPool(
            pool_name="db_pool",
            pool_size=5,
            pool_reset_session=True,
            host=os.getenv("DB_HOST"),  # 數(shù)據(jù)庫(kù)服務(wù)器地址
            user=os.getenv("DB_USER"),  # 數(shù)據(jù)庫(kù)用戶名
            password=os.getenv("DB_PASSWD"),  # 數(shù)據(jù)庫(kù)密碼
            database=os.getenv("DB_NAME"),  # 數(shù)據(jù)庫(kù)名
        )

    @contextmanager
    def get_cursor(self):
        with self.connection_pool.get_connection() as connection:
            cursor = None
            try:
                cursor = connection.cursor()
                yield cursor
                connection.commit()
            except Exception as e:
                connection.rollback()
                raise e
            finally:
                if cursor:
                    cursor.close()

    def execute_sql(self, sql: str) -> str:
        with self.get_cursor() as cursor:
            cursor.execute(sql)
            if cursor.description is not None:
                rows = cursor.fetchall()
                result = {
                    "columns": [desc[0] for desc in cursor.description],
                    "rows": rows,
                }
                return json.dumps(result, default=str)
            else:
                return f"row affected:{cursor.rowcount}"


dbManager = DbManager()
# Initialize FastMCP server
mcp = FastMCP()


@mcp.tool()
def get_table_definition(table: str) -> str:
    """get table definition"""
    return dbManager.execute_sql(f"show create table {table}")


@mcp.tool()
def execute_sql(sql: str) -> str:
    """execute sql"""
    return dbManager.execute_sql(sql)


if __name__ == "__main__":
    # Initialize and run the server
    mcp.run(transport="sse")

uv run main.py

{"finance_server":{"url":"http://host.docker.internal:8000/sse","headers":{},"timeout":50,"sse_read_timeout":50}}

# 角色
你是記賬助手，可以通過調(diào)用MCP工具完成記錄日常收入和支出并作分析。
為了獲得MCP工具列表，必須先通過mcp_sse_list_tools獲取。
為了完成記賬操作，需要先獲取表finance的定義。
記賬的用戶ID取值為{{#sys.user_id#}}

# 收支類別
收入：工資薪金，勞務(wù)報(bào)酬，投資收益，分紅收入，租金收入，其它收入
支出：住房，交通，通訊，保險(xiǎn)，餐飲，電子產(chǎn)品，日用品，服飾，旅行，娛樂，醫(yī)療，學(xué)習(xí)，其它支出

# 技能
## 技能1：記錄日常開支
將開支信息記錄到數(shù)據(jù)庫(kù)表finance

## 技能2：統(tǒng)計(jì)日常開支
根據(jù)用戶輸入信息分析統(tǒng)計(jì)日常開支

# 限制
僅處理記賬相關(guān)問題，不回復(fù)其它問題