基于Dify构建公司内部数据问答系统

version: '3.8'

services:
  mysql:
    image: mysql:8.0.43
    container_name: mysql_container
    environment:
      MYSQL_ROOT_PASSWORD: 123456abc
      MYSQL_DATABASE: function_calls
      MYSQL_USER: dev
      MYSQL_PASSWORD: 123456
    ports:
      - "3306:3306"
    command: --local-infile=1
    volumes:
      - ./mysql_data:/var/lib/mysql

volumes:
  mysql_data:

docker compose up -d 

CREATE TABLE `emp` (
  `empno` int DEFAULT NULL, --员工编号, 默认为空
  `ename` varchar(50) DEFAULT NULL, --员工姓名, 默认为空
  `job` varchar(50) DEFAULT NULL,--员工工作, 默认为空
  `mgr` int DEFAULT NULL,--员工领导, 默认为空
  `hiredate` date DEFAULT NULL,--员工入职日期, 默认为空
  `sal` int DEFAULT NULL,--员工的月薪, 默认为空
  `comm` int DEFAULT NULL,--员工年终奖, 默认为空
  `deptno` int DEFAULT NULL,--员工部分编号, 默认为空
);

CREATE TABLE `DEPT` (
  `DEPTNO` int NOT NULL, -- 部门编码, 默认为空
  `DNAME` varchar(14) DEFAULT NULL,--部门名称, 默认为空
  `LOC` varchar(13) DEFAULT NULL,--地点, 默认为空
  PRIMARY KEY (`DEPTNO`)
);

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Author  : Vincent
@Time    : 2026/1/20 16:41
@File    : mcp_server_mysql.py
@Function :
"""
import os

import pymysql
from fastmcp import FastMCP

mcp = FastMCP(name="mysql_server")


class ASKDatabase(object):
    def __init__(self):
        self.conn = pymysql.connect(
            host='localhost',
            port=3306,
            user='root',
            password=os.getenv("PASSWORD"),
            database='function_calls',
            charset='utf8mb4',
        )
        self.cursor = self.conn.cursor()

    def ask_database(self, query):
        """进行查询"""
        self.cursor.execute(query)
        return self.cursor.fetchall()

    def close(self):
        self.cursor.close()
        self.conn.close()


db = ASKDatabase()


@mcp.tool(
    name="query_mysql_question",
    description="从mysql数据库查询公司的人员详细数据"
)
def query_mysql(query: str):
    """
    query_mysql 需要一个参数 query，表示要查询的 SQL 查询语句。
    返回查询结果。
    """
    # 异常处理
    return db.ask_database(query)


def main():
    """
    启动 Streamable HTTP 服务器。
    """
    print("正在启动MCP Streamable服务器...")
    print("服务器将在 http://localhost:5122 上运行")
    print("按 Ctrl+C 停止服务器")

    try:
        # mcp.run(transport="streamable-http")  # 使用 streamable-http 传输方式
        mcp.run(transport="http", host="0.0.0.0", port=5122)
    except KeyboardInterrupt:
        print("\n服务器已停止")
    except Exception as e:
        print(f"服务器启动失败: {e}")


if __name__ == "__main__":
    main()

请根据用户的问题，选择合适的工具或者查询数据库回答用户的问题。

## 能力

### sql语句生成
如果用户咨询员工的基本信息，请跟据下面的mysql的schema字段，生成对应的查询语句，然后调用mysql查询的工具

"""
CREATE TABLE `emp` (
  `empno` int DEFAULT NULL, --员工编号, 默认为空
  `ename` varchar(50) DEFAULT NULL, --员工姓名, 默认为空
  `job` varchar(50) DEFAULT NULL,--员工工作, 默认为空
  `mgr` int DEFAULT NULL,--员工领导, 默认为空
  `hiredate` date DEFAULT NULL,--员工入职日期, 默认为空
  `sal` int DEFAULT NULL,--员工的月薪, 默认为空
  `comm` int DEFAULT NULL,--员工年终奖, 默认为空
  `deptno` int DEFAULT NULL,--员工部分编号, 默认为空
);

CREATE TABLE `DEPT` (
  `DEPTNO` int NOT NULL, -- 部门编码, 默认为空
  `DNAME` varchar(14) DEFAULT NULL,--部门名称, 默认为空
  `LOC` varchar(13) DEFAULT NULL,--地点, 默认为空
  PRIMARY KEY (`DEPTNO`)
);
"""

例如 “公司里员工工资最高的是谁？”

请根据上面的语句输出生成查询sql，查询应该以纯文本返回，而不是JSON。查询应该只包含MySQL支持的语法。
后端使用的mysql的版本为8.0.43，请注意生成sql语句一定要兼容mysql的版本

### 公司员工手册查询
如果用户查询员工手册的信息，例如查询“薪资福利制度”， “请假流程”等信息。
需要查询知识库，然后根据知识库检索的内容回答用户的问题

## 输出要求
请记住，如果是生成sql语句，你只能生成查询语句，如果要求生成除了查询功能以后的语句，请告诉用户“此功能无法实现。”