Dify + Langbot 构建飞书智能机器人

使用Dify构建出智能体，通过LangBot与飞书对接，将智能体以飞书机器人的形式接入群聊，实现在群中可直接使用智能体，方便又好用。LangBot作为中间层，作为联通dify与飞书的工具，大概架构如下：

一、平台搭建与配置

1.1 Dify平台搭建

本地部署可以搭建社区版，GitHub首页：https://github.com/langgenius/dify/

使用docker部署，下载源码：

git clone https://github.com/langgenius/dify.git
cd dify
cd docker
cp .env.example .env
docker compose up -d

启动成功后，访问：http://服务器IP，进入平台，首次输入的账号密码则为管理员账号

云服务器需要放通安全组：80端口

1.2 LangBot平台搭建

GitHub首页：https://github.com/langbot-app/LangBot

使用docker部署

git clone https://gitcode.com/RockChinQ/LangBot
cd LangBot
docker compose up -d

启动成功后，访问：http://服务器IP:5300 ，进入平台，首次输入的账号密码则为管理员账号

云服务器需要放通安全组：5300端口、2280-2290端口供使用 OneBot 协议的消息平台适配器反向连接

二、Dify智能体创建

2.1 对接大模型

登录dify控制台后，点击右上角头像，点设置，进入设置界面后，点击模型供应商，通过安装模型供应商的插件，可以对接常用的商用大模型。例如：月之暗面、阿里百炼、OpenRouter等。可以配置多个，以防止某个厂商的token用完。

配置模型主要使用厂商提供的api key，我这里以阿里百炼为例，登录百炼平台的控制台：https://bailian.console.aliyun.com/

登录后，找的左侧菜单的“密钥管理”，如果没有则创建一个，复制出API Key。

回到Dify平台，点击模型供应商插件卡片后面的配置，然后输入刚刚获取的API Key，等待指示灯变绿，即配置模型成功。

2.2 对接MCP服务

目前dify可以对接sse和streamable http两种远程协议的MCP Server，这里我们可以在服务器上起一个自己开发的ssh mcp server，一段简单的 python 代码，可以实现登录至服务器，并执行命令。

import os
import paramiko
from typing import List, Dict, Any, Optional
from mcp.server.fastmcp import FastMCP
from getpass import getpass
import keyring
import logging

# 设置日志记录
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("ssh_mcp")

# 创建MCP服务器实例
mcp = FastMCP("ssh-server", host="0.0.0.0", port=8001)

# 存储SSH连接信息的字典
ssh_connections = {}

class SSHClient:
    """SSH客户端封装类，支持自定义端口"""
    def __init__(self, hostname: str, username: str, port: int = 22,
                 password: str = None, key_filename: str = None):
        self.hostname = hostname
        self.username = username
        self.port = port
        self.client = paramiko.SSHClient()
        self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

        try:
            connect_params = {
                'hostname': hostname,
                'username': username,
                'port': port,  # 添加端口参数
            }

            if key_filename:
                connect_params['key_filename'] = key_filename
                logger.info(f"尝试使用密钥文件连接到 {hostname}:{port}")
            elif password:
                connect_params['password'] = password
                logger.info(f"尝试使用密码连接到 {hostname}:{port}")
            else:
                raise ValueError("需要密码或密钥文件进行认证")

            self.client.connect(**connect_params)
            logger.info(f"成功连接到服务器 {hostname}:{port}")

        except Exception as e:
            logger.error(f"连接到 {hostname}:{port} 失败: {str(e)}")
            raise

    def execute_command(self, command: str) -> Dict[str, Any]:
        """在远程服务器上执行命令"""
        try:
            stdin, stdout, stderr = self.client.exec_command(command)
            exit_status = stdout.channel.recv_exit_status()

            return {
                "output": stdout.read().decode().strip(),
                "error": stderr.read().decode().strip(),
                "exit_status": exit_status
            }
        except Exception as e:
            return {"error": str(e), "output": "", "exit_status": -1}

    def close(self):
        """关闭SSH连接"""
        self.client.close()
        logger.info(f"关闭与服务器 {self.hostname}:{self.port} 的连接")

@mcp.tool()
def connect_to_server(server_alias: str, hostname: str, username: str, port: int = 22) -> str:
    """
    连接到SSH服务器并保存连接

    Args:
        server_alias: 服务器别名（用于后续引用）
        hostname: 服务器主机名或IP地址
        username: SSH用户名
        port: SSH端口号 (默认: 22)

    Returns:
        连接状态消息
    """
    try:
        # 安全地获取密码
        password = getpass(f"请输入服务器 {hostname}:{port} 的密码: ")

        # 创建SSH连接
        ssh_client = SSHClient(hostname, username, port, password)

        # 存储连接
        ssh_connections[server_alias] = ssh_client

        return f"成功连接到服务器 {hostname}:{port}，别名为: {server_alias}"
    except Exception as e:
        return f"连接失败: {str(e)}"

@mcp.tool()
def connect_to_server_with_password(server_alias: str, hostname: str, username: str,
                                   password: str, port: int = 22) -> str:
    """
    使用密码连接到SSH服务器并保存连接

    Args:
        server_alias: 服务器别名（用于后续引用）
        hostname: 服务器主机名或IP地址
        username: SSH用户名
        password: SSH密码
        port: SSH端口号 (默认: 22)

    Returns:
        连接状态消息
    """
    try:
        # 创建SSH连接
        ssh_client = SSHClient(hostname, username, port, password)

        # 存储连接
        ssh_connections[server_alias] = ssh_client

        return f"成功连接到服务器 {hostname}:{port}，别名为: {server_alias}"
    except Exception as e:
        return f"连接失败: {str(e)}"

@mcp.tool()
def connect_with_key(server_alias: str, hostname: str, username: str,
                    key_path: str, port: int = 22) -> str:
    """
    使用SSH密钥连接到服务器

    Args:
        server_alias: 服务器别名
        hostname: 服务器主机名或IP
        username: SSH用户名
        key_path: SSH私钥文件路径
        port: SSH端口号 (默认: 22)

    Returns:
        连接状态消息
    """
    try:
        ssh_client = SSHClient(hostname, username, port, key_filename=key_path)
        ssh_connections[server_alias] = ssh_client
        return f"成功使用密钥连接到服务器 {hostname}:{port}"
    except Exception as e:
        return f"连接失败: {str(e)}"

@mcp.tool()
def execute_remote_command(server_alias: str, command: str) -> str:
    """
    在指定的远程服务器上执行命令

    Args:
        server_alias: 之前连接的服务器别名
        command: 要执行的Linux命令

    Returns:
        命令执行结果
    """
    if server_alias not in ssh_connections:
        return f"错误: 未找到服务器别名 '{server_alias}'，请先使用connect_to_server连接"

    try:
        ssh_client = ssh_connections[server_alias]
        result = ssh_client.execute_command(command)

        # 在结果中显示服务器地址和端口
        server_info = f"{ssh_client.hostname}:{ssh_client.port}"

        if result["exit_status"] == 0:
            return f"服务器 {server_info} 命令执行成功:\n{result['output']}"
        else:
            return f"服务器 {server_info} 命令执行失败 (退出码: {result['exit_status']}):\n错误: {result['error']}\n输出: {result['output']}"
    except Exception as e:
        return f"执行命令时发生错误: {str(e)}"

@mcp.tool()
def list_connected_servers() -> str:
    """
    列出当前所有已连接的服务器别名及其连接信息

    Returns:
        已连接服务器列表
    """
    if not ssh_connections:
        return "当前没有已连接的服务器"

    servers_list = "已连接的服务器:\n"
    for alias, client in ssh_connections.items():
        servers_list += f"- {alias} ({client.hostname}:{client.port}, 用户: {client.username})\n"

    return servers_list

@mcp.tool()
def disconnect_server(server_alias: str) -> str:
    """
    断开与指定服务器的连接

    Args:
        server_alias: 要断开的服务器别名

    Returns:
        断开连接的状态消息
    """
    if server_alias not in ssh_connections:
        return f"错误: 未找到服务器别名 '{server_alias}'"

    try:
        ssh_client = ssh_connections[server_alias]
        server_info = f"{ssh_client.hostname}:{ssh_client.port}"
        ssh_client.close()
        del ssh_connections[server_alias]
        return f"已断开与服务器 '{server_alias}' ({server_info}) 的连接"
    except Exception as e:
        return f"断开连接时发生错误: {str(e)}"

@mcp.tool()
def change_server_port(server_alias: str, new_port: int) -> str:
    """
    修改已连接服务器的端口配置（需要重新连接）

    Args:
        server_alias: 服务器别名
        new_port: 新的SSH端口号

    Returns:
        修改结果消息
    """
    if server_alias not in ssh_connections:
        return f"错误: 未找到服务器别名 '{server_alias}'"

    try:
        ssh_client = ssh_connections[server_alias]
        old_port = ssh_client.port
        ssh_client.port = new_port
        return f"已将服务器 '{server_alias}' 的端口从 {old_port} 修改为 {new_port}。注意：需要重新连接才能生效。"
    except Exception as e:
        return f"修改端口时发生错误: {str(e)}"

@mcp.tool()
def test_ssh_port(hostname: str, port: int = 22, timeout: int = 5) -> str:
    """
    测试SSH端口的连通性

    Args:
        hostname: 服务器主机名或IP地址
        port: 要测试的SSH端口号 (默认: 22)
        timeout: 连接超时时间（秒）

    Returns:
        端口测试结果
    """
    import socket

    try:
        # 创建socket连接测试端口
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(timeout)
        result = sock.connect_ex((hostname, port))
        sock.close()

        if result == 0:
            return f"SSH端口 {hostname}:{port} 可达 ✓"
        else:
            return f"SSH端口 {hostname}:{port} 不可达 ✗ (错误码: {result})"
    except Exception as e:
        return f"测试端口时发生错误: {str(e)}"

if __name__ == "__main__":
    mcp.run(transport="streamable-http")

可以使用 uv 命令启动该mcp server

# 安装uv命令
pip install uv
# 创建mcp server目录
mkdir ssh-command-server
cd ssh-command-server
# 初始化
uv init .
uv add "mcp[cli]"
# 将上面代码写入main.py文件中，启动server
nohup uv run main.py >> app.log 2>&1 &

启动之后，该mcp server 的访问方式为：http://服务器IP:8001/mcp

回到dify控制台后，点击“工具”，点击“MCP”，点击“添加 MCP 服务 (HTTP)”，添加一个MCP Server。

注意：如果是sse协议，服务端点则改成：http://服务器IP:8001/sse

添加成功后的效果，可以看到该server下的所有工具，也就是python代码里写的方法。

2.3 创建智能体

回到dify控制台后，点击“工作室”，点击“创建空白应用”，展开“新手适用”，点击“Agent”，起一个名字后，创建。

进入到编排页面，填写“提示词”、添加工具，MCP Server中的工具全部加进去，选择大模型，我这里选百炼提供的DeepSeek-v3。测试没问题后，点击发布。

发布完之后，点击左侧的”访问API”，复制右上角的API 服务器地址，并且创建一个API密钥。保存好API地址与密钥，用于LangBot对接。

三、飞书机器人创建

登录飞书开发者后台：https://open.feishu.cn/app?lang=zh-CN，创建一个”企业自建应用“，输入应用名称和应用描述后创建一个应用。添加应用能力选择“机器人”。接下来是对该应用的配置：

3.1 复制凭证

点击左侧菜单”凭证与基础信息“。复制保存应用凭证的”App ID“与”App Secret“，用于LangBot对接。

3.2 权限管理

点击左侧菜单”权限管理“，添加开通以下权限：cardkit:card:read、cardkit:card:write、im:message、im:message.group_at_msg:readonly、im:message.p2p_msg:readonly

3.3 事件与回调

点击左侧菜单”事件与回调“，”事件配置“订阅方式选择”长连接“。

添加”接收消息“事件，并选择相关权限：

3.4 发布版本

根据提示创建版本，后点击发布，飞书机器人应用正式发布。

发布之后，在飞书群组中，点击添加机器人，搜索刚刚发布的机器人名称，添加到群组。

四、LangBot机器人对接

4.1 添加机器人

登录LangBot控制台，点击左侧”机器人“菜单，添加机器人。平台/适配器选择”飞书“，应用ID、应用密钥、机器人名称一定要与飞书平台创建的一致。提交之后点击保存，其他选项先默认。

4.2 添加流水线

点击左侧”流水线“菜单，默认已经存在一个ChatPipeline，作为所有机器人的默认流水线。

我们添加一个新的流水线，用于对接dify。运行器选择”Dify服务API“，基础URL和API密钥贴入dify智能体发布后复制的内容。

4.3 修改机器流水线

返回机器人，修改机器人的流水线为刚刚对接的dify流水线，点击保存

五、测试效果

接下来可以通过@群内的机器人来调用dify的智能体进行对话了。