登录/ 注册
 找回密码
 立即注册
关闭

CSDN热搜

程序园 精品问答 技术交流 资源下载
返回列表 发新帖
首页 › 业界区 › 业界 ›dify+MCP多应用，构建灵活的AI应用生态系统 ...

dify+MCP多应用，构建灵活的AI应用生态系统

丝甲坞 2025-6-3 00:01:55
一、概述

前面几篇文章写很多MCP应用，基本上一个dify工作流使用一个MCP应用。
那么一个dify工作流，同时使用多个MCP应用，是否可以呢？答案是可以的。
 
先来看一下效果图
1.png

说明：
这里使用了问题分类器，用来判断用户的问题，应该调用哪个MCP应用
AGENT1~4，分别对应一个MCP应用，例如：public-ip-mcp-server，mysql8-mcp-server，desensitization-mcp-server，searxng-mcp-server
针对mysql查询输出的内容，会进行脱敏处理。
二、问题分类器

定义

通过定义分类描述，问题分类器能够根据用户输入，使用 LLM 推理与之相匹配的分类并输出分类结果，向下游节点提供更加精确的信息。
场景

常见的使用情景包括客服对话意图分类、产品评价分类、邮件批量分类等。
在一个典型的产品客服问答场景中，问题分类器可以作为知识库检索的前置步骤，对用户输入问题意图进行分类处理，分类后导向下游不同的知识库查询相关的内容，以精确回复用户的问题。
设置

对于比较精确的条件，一般使用条件分支。但是对于我这种场景，条件比较模糊，所以需要使用问题分类器
 
这里定义了3个分类：
  1. 公网ip相关问题
  2. mysql 数据库相关查询，涉及学生、教师、成绩、班级、课程等
  3. 其他问题
复制代码
效果如下：
2.png

 说明：
公网ip相关问题，会直接调用MCP应用public-ip-mcp-server
mysql相关问题，会调用MCP应用mysql8-mcp-server
其他问题，会调用MCP应用searxng-mcp-server，这个是一个联网搜索引擎，你可以理解为百度，想搜什么都可以。
三、环境说明

dify版本

这里使用的是最新版本1.4.0，如果你的版本没有这么高，1.3.0以上版本也可以。
mcp插件

确保已经安装了以下插件：
  1. Agent 策略(支持 MCP 工具)
  3. MCP SSE / StreamableHTTP
复制代码
确保插件版本，已经升级到最新版本
mcp应用

这里的所有MCP应用，统一使用Streamable HTTP模式，全部部署在k8s里面。
当然，使用docker运行也是可以的。
mcp插件设置

点击插件MCP SSE / StreamableHTTP，输入MCP 服务配置
3.png

 完整内容如下：
  1. {
  2.     "mysql8-mcp-server": {
  3.         "transport": "streamable_http",
  4.         "url": "http://mysql8-mcp-server-svc.mcp:9000/mcp/",
  5.         "headers": {},
  6.         "timeout": 60
  7.     },
  8.     "desensitization-mcp-server": {
  9.         "transport": "streamable_http",
  10.         "url": "http://desensitization-mcp-server-svc.mcp:9000/mcp/",
  11.         "headers": {},
  12.         "timeout": 60
  13.     },
  14.     "public-ip-mcp-server": {
  15.         "transport": "streamable_http",
  16.         "url": "http://public-ip-mcp-server-svc.mcp:9000/mcp/",
  17.         "headers": {},
  18.         "timeout": 60
  19.     },
  20.     "searxng-mcp-server": {
  21.         "transport": "streamable_http",
  22.         "url": "http://searxng-mcp-server-svc.mcp:9000/mcp/",
  23.         "headers": {},
  24.         "timeout": 60
  25.     }
  26. }
复制代码
注意：这里的url使用的是k8s内部地址，如果使用的是docker方式运行，请根据实际情况修改。
四、public-ip-mcp-server设置

public-ip-mcp-server核心代码如下：
server.py
4.gif
5.gif
  1. from fastmcp import FastMCP
  2. import json
  3. import requests
  5. mcp = FastMCP("public-ip-address")
  8. @mcp.tool()
  9. def get_public_ip_address() -> str:
  10.     """
  11.     获取公网ip地址
  12.     返回:
  13.         str: 当前网络的公网ip地址
  14.     """
  15.     response = requests.get("http://ip-api.com/json")
  16.     content = json.loads(response.text)
  17.     return content["query"]
  20. if __name__ == "__main__":
  21.     mcp.run(transport="streamable-http", host="0.0.0.0", port=9000, path="/mcp")
复制代码
View CodeAgent配置

Agent 1详细配置如下：
6.png

 MCP服务配置
  1. {
  2.   "public-ip-mcp-server": {
  3.         "transport": "streamable_http",
  4.         "url": "http://public-ip-mcp-server-svc.mcp:9000/mcp/",
  5.         "headers": {},
  6.         "timeout": 60
  7.     }
  8. }
复制代码
指令
  1. 使用MCP工具，获取服务器公网ip
复制代码
 
最后直接回复，注意选择变量Agent1 text
7.png

 
五、mysql8-mcp-server设置

核心代码

mysql8-mcp-server核心代码如下：
server.py
8.gif
9.gif
  1. from fastmcp import FastMCP
  2. from mysql.connector import connect, Error
  3. import os
  5. mcp = FastMCP("operateMysql")
  8. def get_db_config():
  9.     """从环境变量获取数据库配置信息
  11.     返回:
  12.         dict: 包含数据库连接所需的配置信息
  13.         - host: 数据库主机地址
  14.         - port: 数据库端口
  15.         - user: 数据库用户名
  16.         - password: 数据库密码
  17.         - database: 数据库名称
  19.     异常:
  20.         ValueError: 当必需的配置信息缺失时抛出
  21.     """
  23.     config = {
  24.         "host": os.getenv("MYSQL_HOST", "localhost"),
  25.         "port": int(os.getenv("MYSQL_PORT", "3306")),
  26.         "user": os.getenv("MYSQL_USER"),
  27.         "password": os.getenv("MYSQL_PASSWORD"),
  28.         "database": os.getenv("MYSQL_DATABASE"),
  29.     }
  30.     print(config)
  31.     if not all(
  32.         [
  33.             config["host"],
  34.             config["port"],
  35.             config["user"],
  36.             config["password"],
  37.             config["database"],
  38.         ]
  39.     ):
  40.         raise ValueError("缺少必需的数据库配置")
  42.     return config
  45. @mcp.tool()
  46. def execute_sql(query: str) -> list:
  47.     """执行SQL查询语句
  49.     参数:
  50.         query (str): 要执行的SQL语句，支持多条语句以分号分隔
  52.     返回:
  53.         list: 包含查询结果的TextContent列表
  54.         - 对于SELECT查询：返回CSV格式的结果，包含列名和数据
  55.         - 对于SHOW TABLES：返回数据库中的所有表名
  56.         - 对于其他查询：返回执行状态和影响行数
  57.         - 多条语句的结果以"---"分隔
  59.     异常:
  60.         Error: 当数据库连接或查询执行失败时抛出
  61.     """
  62.     config = get_db_config()
  63.     try:
  64.         with connect(**config) as conn:
  65.             with conn.cursor() as cursor:
  66.                 statements = [stmt.strip() for stmt in query.split(";") if stmt.strip()]
  67.                 results = []
  69.                 for statement in statements:
  70.                     try:
  71.                         cursor.execute(statement)
  73.                         # 检查语句是否返回了结果集 (SELECT, SHOW, EXPLAIN, etc.)
  74.                         if cursor.description:
  75.                             columns = [desc[0] for desc in cursor.description]
  76.                             rows = cursor.fetchall()
  78.                             # 将每一行的数据转换为字符串，特殊处理None值
  79.                             formatted_rows = []
  80.                             for row in rows:
  81.                                 formatted_row = [
  82.                                     "NULL" if value is None else str(value)
  83.                                     for value in row
  84.                                 ]
  85.                                 formatted_rows.append(",".join(formatted_row))
  87.                             # 将列名和数据合并为CSV格式
  88.                             results.append(
  89.                                 "\n".join([",".join(columns)] + formatted_rows)
  90.                             )
  92.                         # 如果语句没有返回结果集 (INSERT, UPDATE, DELETE, etc.)
  93.                         else:
  94.                             conn.commit()  # 只有在非查询语句时才提交
  95.                             results.append(f"查询执行成功。影响行数: {cursor.rowcount}")
  97.                     except Error as stmt_error:
  98.                         # 单条语句执行出错时，记录错误并继续执行
  99.                         results.append(
  100.                             f"执行语句 '{statement}' 出错: {str(stmt_error)}"
  101.                         )
  102.                         # 可以在这里选择是否继续执行后续语句，目前是继续
  104.                 return ["\n---\n".join(results)]
  106.     except Error as e:
  107.         print(f"执行SQL '{query}' 时出错: {e}")
  108.         return [f"执行查询时出错: {str(e)}"]
  111. @mcp.tool()
  112. def get_table_name(text: str) -> list:
  113.     """根据表的中文注释搜索数据库中的表名
  115.     参数:
  116.         text (str): 要搜索的表中文注释关键词
  118.     返回:
  119.         list: 包含查询结果的TextContent列表
  120.         - 返回匹配的表名、数据库名和表注释信息
  121.         - 结果以CSV格式返回，包含列名和数据
  122.     """
  123.     config = get_db_config()
  124.     sql = "SELECT TABLE_SCHEMA, TABLE_NAME, TABLE_COMMENT "
  125.     sql += f"FROM information_schema.TABLES WHERE TABLE_SCHEMA = '{config['database']}' AND TABLE_COMMENT LIKE '%{text}%';"
  126.     return execute_sql(sql)
  129. @mcp.tool()
  130. def get_table_desc(text: str) -> list:
  131.     """获取指定表的字段结构信息
  133.     参数:
  134.         text (str): 要查询的表名，多个表名以逗号分隔
  136.     返回:
  137.         list: 包含查询结果的列表
  138.         - 返回表的字段名、字段注释等信息
  139.         - 结果按表名和字段顺序排序
  140.         - 结果以CSV格式返回，包含列名和数据
  141.     """
  142.     config = get_db_config()
  143.     # 将输入的表名按逗号分割成列表
  144.     table_names = [name.strip() for name in text.split(",")]
  145.     # 构建IN条件
  146.     table_condition = "','".join(table_names)
  147.     sql = "SELECT TABLE_NAME, COLUMN_NAME, COLUMN_COMMENT "
  148.     sql += (
  149.         f"FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = '{config['database']}' "
  150.     )
  151.     sql += f"AND TABLE_NAME IN ('{table_condition}') ORDER BY TABLE_NAME, ORDINAL_POSITION;"
  152.     return execute_sql(sql)
  155. @mcp.tool()
  156. def get_lock_tables() -> list:
  157.     """
  158.     获取当前mysql服务器InnoDB 的行级锁
  160.     返回:
  161.         list: 包含查询结果的TextContent列表
  162.     """
  163.     sql = """SELECT
  164.     p2.`HOST` AS 被阻塞方host,
  165.     p2.`USER` AS 被阻塞方用户,
  166.     r.trx_id AS 被阻塞方事务id,
  167.     r.trx_mysql_thread_id AS 被阻塞方线程号,
  168.     TIMESTAMPDIFF(SECOND, r.trx_wait_started, CURRENT_TIMESTAMP) AS 等待时间,
  169.     r.trx_query AS 被阻塞的查询,
  170.     l.OBJECT_NAME AS 阻塞方锁住的表,
  171.     m.LOCK_MODE AS 被阻塞方的锁模式,
  172.     m.LOCK_TYPE AS '被阻塞方的锁类型(表锁还是行锁)',
  173.     m.INDEX_NAME AS 被阻塞方锁住的索引,
  174.     m.OBJECT_SCHEMA AS 被阻塞方锁对象的数据库名,
  175.     m.OBJECT_NAME AS 被阻塞方锁对象的表名,
  176.     m.LOCK_DATA AS 被阻塞方事务锁定记录的主键值,
  177.     p.`HOST` AS 阻塞方主机,
  178.     p.`USER` AS 阻塞方用户,
  179.     b.trx_id AS 阻塞方事务id,
  180.     b.trx_mysql_thread_id AS 阻塞方线程号,
  181.     b.trx_query AS 阻塞方查询,
  182.     l.LOCK_MODE AS 阻塞方的锁模式,
  183.     l.LOCK_TYPE AS '阻塞方的锁类型(表锁还是行锁)',
  184.     l.INDEX_NAME AS 阻塞方锁住的索引,
  185.     l.OBJECT_SCHEMA AS 阻塞方锁对象的数据库名,
  186.     l.OBJECT_NAME AS 阻塞方锁对象的表名,
  187.     l.LOCK_DATA AS 阻塞方事务锁定记录的主键值,
  188.     IF(p.COMMAND = 'Sleep', CONCAT(p.TIME, ' 秒'), 0) AS 阻塞方事务空闲的时间
  189.     FROM performance_schema.data_lock_waits w
  190.     INNER JOIN performance_schema.data_locks l ON w.BLOCKING_ENGINE_LOCK_ID = l.ENGINE_LOCK_ID
  191.     INNER JOIN performance_schema.data_locks m ON w.REQUESTING_ENGINE_LOCK_ID = m.ENGINE_LOCK_ID
  192.     INNER JOIN information_schema.INNODB_TRX b ON b.trx_id = w.BLOCKING_ENGINE_TRANSACTION_ID
  193.     INNER JOIN information_schema.INNODB_TRX r ON r.trx_id = w.REQUESTING_ENGINE_TRANSACTION_ID
  194.     INNER JOIN information_schema.PROCESSLIST p ON p.ID = b.trx_mysql_thread_id
  195.     INNER JOIN information_schema.PROCESSLIST p2 ON p2.ID = r.trx_mysql_thread_id
  196.     ORDER BY 等待时间 DESC;"""
  198.     return execute_sql(sql)
  201. if __name__ == "__main__":
  202.     mcp.run(transport="streamable-http", host="0.0.0.0", port=9000, path="/mcp")
复制代码
View CodeAgent配置

Agent 2详细配置如下：
10.png

 
11.png

 MCP服务配置
  1. {
  2.     "mysql8-mcp-server": {
  3.         "transport": "streamable_http",
  4.         "url": "http://mysql8-mcp-server-svc.mcp:9000/mcp/",
  5.         "timeout": 60
  6.     }
  7. }
复制代码
指令
12.gif
13.gif
  1. 使用中文回复。
  3. 当用户提问中涉及学生、教师、成绩、班级、课程等实体时，需要使用 MySQL MCP 进行数据查询和操作，表结构说明如下：
  5. # 学生管理系统数据库表结构说明
  7. ## 1. 教师表 (teachers)
  9. | 字段名 | 类型 | 描述 | 约束 | 示例 |
  10. |--------|------|------|------|------|
  11. | id | varchar | 教师ID | 主键 | "T001" |
  12. | name | varchar | 教师姓名 | 必填 | "张建国" |
  13. | gender | enum | 性别 | "男"或"女" | "男" |
  14. | subject | varchar | 教授科目 | 必填 | "数学" |
  15. | title | varchar | 职称 | 必填 | "教授" |
  16. | phone | varchar | 联系电话 | 必填 | "13812345678" |
  17. | office | varchar | 办公室位置 | 必填 | "博学楼301" |
  18. | wechat | varchar | 微信(可选) | 可选 | "lily_teacher" |
  19. | isHeadTeacher | enum | 是否为班主任，"true"或"false" | 可选 | true |
  21. ## 2. 班级表 (classes)
  23. | 字段名 | 类型 | 描述 | 约束 | 示例 |
  24. |--------|------|------|------|------|
  25. | id | varchar | 班级ID | 主键 | "202301" |
  26. | className | varchar | 班级名称 | 必填 | "2023级计算机1班" |
  27. | grade | int | 年级 | 必填 | 2023 |
  28. | headTeacherId | varchar | 班主任ID | 外键(teachers.id) | "T003" |
  29. | classroom | varchar | 教室位置 | 必填 | "1号楼302" |
  30. | studentCount | int | 学生人数 | 必填 | 35 |
  31. | remark | varchar | 备注信息 | 可选 | "市级优秀班集体" |
  33. ## 3. 课程表 (courses)
  35. | 字段名 | 类型 | 描述 | 约束 | 示例 |
  36. |--------|------|------|------|------|
  37. | id | varchar | 课程ID | 主键 | "C001" |
  38. | courseName | varchar | 课程名称 | 必填 | "高等数学" |
  39. | credit | int | 学分 | 必填 | 4 |
  40. | teacherId | varchar | 授课教师ID | 外键(teachers.id) | "T001" |
  41. | semester | varchar | 学期 | 格式"YYYY-N" | "2023-1" |
  42. | type | enum | 课程类型 | "必修"或"选修" | "必修" |
  43. | prerequisite | varchar | 先修课程ID | 可选,外键(courses.id) | "C003" |
  45. ## 4. 学生表 (students)
  47. | 字段名 | 类型 | 描述 | 约束 | 示例 |
  48. |--------|------|------|------|------|
  49. | id | varchar | 学号 | 主键 | "S20230101" |
  50. | name | varchar | 学生姓名 | 必填 | "王强" |
  51. | gender | enum | 性别 | "男"或"女" | "男" |
  52. | birthDate | date | 出生日期 | 必填 | date("2005-01-15") |
  53. | enrollmentDate | date | 入学日期 | 必填 | date("2023-8-1") |
  54. | classId | varchar | 班级ID | 外键(classes.id) | "202301" |
  55. | phone | varchar | 联系电话 | 必填 | "13812345678" |
  56. | email | varchar | 电子邮箱 | 必填 | "20230101@school.edu.cn" |
  57. | emergencyContact | varchar | 紧急联系人电话 | 必填 | "13876543210" |
  58. | address | varchar | 家庭住址 | 必填 | "北京市海淀区中关村大街1栋101室" |
  59. | height | int | 身高(cm) | 必填 | 175 |
  60. | weight | int | 体重(kg) | 必填 | 65 |
  61. | healthStatus | enum | 健康状况 | 必填，"良好"或"一般‌"或"较差" | "良好" |
  63. ## 5. 成绩表 (scores)
  65. | 字段名 | 类型 | 描述 | 约束 | 示例 |
  66. |--------|------|------|------|------|
  67. | id | varchar | 成绩记录ID | 主键 | "S20230101C001" |
  68. | studentId | varchar | 学生ID | 外键(students.id) | "S20230101" |
  69. | courseId | varchar | 课程ID | 外键(courses.id) | "C001" |
  70. | score | int | 综合成绩 | 0-100 | 85 |
  71. | examDate | date | 考试日期 | 必填 | date("2024-5-20") |
  72. | usualScore | int | 平时成绩 | 0-100 | 90 |
  73. | finalScore | int | 期末成绩 | 0-100 | 80 |
  75. ### 补考成绩记录说明
  76. 补考记录在_id后添加"_M"后缀，如"S20230101C001_M"
  78. ## 表关系说明
  80. 1. **一对多关系**:
  81.    - 一个班级(classes)对应多个学生(students)
  82.    - 一个教师(teachers)可以教授多门课程(courses)
  83.    - 一个学生(students)有多条成绩记录(scores)
  85. 2. **外键约束**:
  86.    - students.classId → classes.id
  87.    - courses.teacherId → teachers.id
  88.    - scores.studentId → students.id
  89.    - scores.courseId → courses.id
  90.    - classes.headTeacherId → teachers.id
复制代码
View Code 
六、desensitization-mcp-server设置

核心代码

desensitization-mcp-server核心代码如下：
server.py
14.gif
15.gif
  1. from fastmcp import FastMCP
  2. import re
  4. mcp = FastMCP("desensitize-text")
  7. class DataMasker:
  8.     def __init__(self):
  9.         pass
  11.     def mask_phone_number(self, phone_number):
  12.         """
  13.         对手机号码进行脱敏处理，将中间四位替换为 *
  14.         """
  15.         if len(phone_number) == 11:
  16.             return phone_number[:3] + "****" + phone_number[7:]
  17.         return phone_number
  19.     def mask_email(self, email):
  20.         """
  21.         对邮箱地址进行脱敏处理，只显示邮箱名前两位和域名
  22.         """
  23.         if "@" in email:
  24.             username, domain = email.split("@")
  25.             return username[:2] + "****@" + domain
  26.         return email
  28.     def mask_id_card(self, id_card):
  29.         """
  30.         对身份证号码进行脱敏处理，只显示前四位和后四位
  31.         """
  32.         if len(id_card) == 18:
  33.             return id_card[:4] + "**********" + id_card[14:]
  34.         return id_card
  36.     def mask_address(self, address):
  37.         """
  38.         对地址进行脱敏处理，模糊化门牌号和房间号
  39.         例如，将 "1栋" 替换为 "**栋"，"101室" 替换为 "***室"
  40.         """
  41.         # 使用正则表达式
  42.         desensitized_address = re.sub(r"(\d+)栋", r"**栋", address)
  43.         desensitized_address = re.sub(r"(\d+)室", r"***室", desensitized_address)
  44.         return desensitized_address
  47. @mcp.tool()
  48. def desensitize_text(text: str) -> str:
  49.     """
  50.     脱敏文本信息
  51.     """
  52.     masker = DataMasker()
  53.     # 匹配手机号
  54.     phone_pattern = r"\d{11}"
  55.     phones = re.findall(phone_pattern, text)
  56.     for phone in phones:
  57.         masked_phone = masker.mask_phone_number(phone)
  58.         text = text.replace(phone, masked_phone)
  60.     # 匹配邮箱
  61.     email_pattern = r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}"
  62.     emails = re.findall(email_pattern, text)
  63.     for email in emails:
  64.         masked_email = masker.mask_email(email)
  65.         text = text.replace(email, masked_email)
  67.     # 匹配身份证号
  68.     id_card_pattern = r"\d{18}"
  69.     id_cards = re.findall(id_card_pattern, text)
  70.     for id_card in id_cards:
  71.         masked_id_card = masker.mask_id_card(id_card)
  72.         text = text.replace(id_card, masked_id_card)
  74.     # 匹配地址
  75.     address_pattern = r"([\u4e00-\u9fa5]+省)?([\u4e00-\u9fa5]+市)?([\u4e00-\u9fa5]+区)?([\u4e00-\u9fa5]+街道)?(\d+[\u4e00-\u9fa5]+)?(\d+[\u4e00-\u9fa5]+)?"
  76.     addresss = re.findall(address_pattern, text)
  78.     # 如果没有找到地址，返回原始文本
  79.     if not addresss:
  80.         return text
  82.     # 对每个匹配的地址进行脱敏处理
  83.     for address_parts in addresss:
  84.         # 将匹配的地址部分组合成完整的地址
  85.         address = "".join([part for part in address_parts if part])
  86.         if address:
  87.             # print("address",address)
  88.             masked_address = masker.mask_address(address)
  89.             text = text.replace(address, masked_address)
  91.     return text
  94. if __name__ == "__main__":
  95.     mcp.run(transport="streamable-http", host="0.0.0.0", port=9000, path="/mcp")
复制代码
View CodeAgent配置

Agent 4详细配置如下：
16.png

注意：查询要选择变量Agent 2 text
 
 MCP服务配置
  1. {
  2.   "desensitization-mcp-server": {
  3.         "transport": "streamable_http",
  4.         "url": "http://desensitization-mcp-server-svc.mcp:9000/mcp/",
  5.         "headers": {},
  6.         "timeout": 60
  7.     }
  8. }
复制代码
指令
  1. 使用MCP工具，对文本进行脱敏处理
复制代码
 
最后直接回复，注意选择变量Agent4 text
17.png

 
七、searxng-mcp-server设置

核心代码

searxng-mcp-server核心代码如下：
server.py
18.gif
19.gif
  1. from fastmcp import FastMCP
  2. import requests
  3. import os
  5. mcp = FastMCP("searxng")
  8. @mcp.tool()
  9. def search(query: str) -> str:
  10.     """
  11.     搜索关键字，调用searxng的API接口
  12.     参数:
  13.         query (str): 要搜索的关键词
  14.     返回:
  15.         str: 查询结果
  16.     """
  17.     api_server = os.getenv("API_SERVER", None)
  18.     if not api_server:
  19.         print("缺少必需的API_SERVER配置")
  20.         raise ValueError("缺少必需的API_SERVER配置")
  22.     # API URL
  23.     url = "%s/search?q=%s&format=json" % (api_server, query)
  24.     print(url)
  26.     try:
  27.         # 发送GET请求
  28.         response = requests.get(url)
  30.         # 检查请求是否成功
  31.         if response.status_code == 200:
  32.             # 将响应内容解析为JSON
  33.             data = response.json()
  34.             # print("JSON内容:")
  35.             # print(data,type(data))
  36.             result_list = []
  37.             for i in data["results"]:
  38.                 # print(i["content"])
  39.                 result_list.append(i["content"])
  40.             content = "\n".join(result_list)
  41.             # print(content)
  42.             return content
  43.         else:
  44.             print(f"请求失败，状态码: {response.status_code}")
  45.             return False
  47.     except requests.exceptions.RequestException as e:
  48.         print(f"请求过程中发生错误: {e}")
  49.         return False
  52. if __name__ == "__main__":
  53.     mcp.run(transport="streamable-http", host="0.0.0.0", port=9000, path="/mcp")
复制代码
View CodeAgent配置

Agent 3详细配置如下：
20.png

 MCP服务配置
  1. {
  2.   "searxng-mcp-server": {
  3.         "transport": "streamable_http",
  4.         "url": "http://searxng-mcp-server-svc.mcp:9000/mcp/",
  5.         "headers": {},
  6.         "timeout": 60
  7.     }
  8. }
复制代码
指令
  1. ## 技能
  2. ### 技能1：使用MCP工具进行联网搜索，获取到的相关内容进行总结分析
  3. ## 限制
  4. - 如果没有相关内容，再进行联网搜索
  5. - 你的回答应严格针对分析任务。使用结构化语言，逐步思考
  6. - 使用的语言应和用户提问的语言相同
  7. - 搜索的关键词必须和用户提问的内容一致
复制代码
 
最后直接回复，注意选择变量Agent3 text
21.png

八、dify测试

点击右上角的预览按钮
22.png

公网ip多少

23.png

注意：这里可以看到绿色的连接线条，可以清晰的看到工作流的走向，它确实是按照我预期的方向在走。
 
李华的老师，查询一下个人详细信息

24.png

可以看到通过问题分类器，分别走向AGENT 2，AGENT 4，最终得到的答案，是进行了脱敏处理。
上海今天天气如何

25.png

这里直接联网搜索答案了
 

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

使用道具 举报
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册
发帖
丝甲坞
2025-6-3 00:01:55

0

粉丝关注

13

主题发布

板块介绍填写区域，请于后台编辑
微信扫一扫