如何不通过Bot Framework Emulator创建Direct Line通道对接本地Bot

Bot Framework Emulator在49780端口提供的本质是一个本地运行的Direct Line协议代理，负责在Direct Line 3.0 API规范和本地Bot识别的Bot Framework Protocol之间做双向转换，完全可以自己实现这个代理跳过模拟器，具体方案如下：

• 本地运行的Python Bot默认暴露/api/messages接口，只识别标准Bot Framework活动格式，本身不直接支持Direct Line协议
• 模拟器的49780端口服务做了三件事：实现Direct Line 3.0的接口供Postman这类客户端调用、把客户端发的Direct Line格式请求转换成Bot能识别的格式转发给3978端口、接收Bot返回的消息存在本地队列供客户端拉取
• 自己实现同等逻辑的代理，就能完全替代模拟器的作用

方式一：自定义兼容Direct Line 3.0的本地代理 #

你可以用任意熟悉的Web框架写代理服务，必须实现的核心接口如下：

• POST /v3/directline/conversations：创建对话，返回随机生成的conversationId、访问token，本地测试不需要做真实的token校验
• POST /v3/directline/conversations/{conversationId}/activities：接收客户端发送的消息，补全serviceUrl、对话双方ID等必填字段后，转发给本地Bot的http://localhost:3978/api/messages接口
• GET /v3/directline/conversations/{conversationId}/activities：返回当前对话下Bot发送的所有未读消息，拉取后标记为已读
• POST /v3/conversations/{conversationId}/activities：供Bot回调发送回复消息，收到Bot的请求后把消息存入对应对话的消息队列即可

方式二：可直接运行的Python极简代理 #

下面的代码用Flask实现了和模拟器完全等价的代理能力，支持创建对话、发消息、拉取活动全流程：

import uuid
import requests
from flask import Flask, request, jsonify
from collections import defaultdict

app = Flask(__name__)
# 本地Bot服务地址
BOT_MSG_ENDPOINT = "http://localhost:3978/api/messages"
# 代理监听端口，和模拟器默认端口一致可直接复用之前的Postman脚本
PROXY_LISTEN_PORT = 49780
# 按对话ID存储消息队列
conv_msg_store = defaultdict(list)
# 存储对话对应的合法token
conv_token_map = {}

@app.post("/v3/directline/conversations")
def create_conv():
    conv_id = str(uuid.uuid4())
    access_token = str(uuid.uuid4())
    conv_token_map[conv_id] = access_token
    return jsonify({
        "conversationId": conv_id,
        "token": access_token,
        "expires_in": 3600,
        "streamUrl": f"ws://localhost:{PROXY_LISTEN_PORT}/v3/directline/conversations/{conv_id}/stream"
    })

@app.post("/v3/directline/conversations/<conv_id>/activities")
def send_msg_to_bot(conv_id):
    req_body = request.get_json()
    # 补全Bot要求的标准活动字段
    req_body["serviceUrl"] = f"http://localhost:{PROXY_LISTEN_PORT}"
    req_body["conversation"] = {"id": conv_id}
    req_body.setdefault("from", {"id": "test_user", "name": "Test User"})
    req_body["recipient"] = {"id": "local_bot", "name": "Local Running Bot"}
    # 转发请求到本地Bot
    requests.post(BOT_MSG_ENDPOINT, json=req_body, headers={"Content-Type": "application/json"})
    return jsonify({"id": str(uuid.uuid4())})

@app.get("/v3/directline/conversations/<conv_id>/activities")
def pull_bot_msg(conv_id):
    # 取出未读消息后清空队列
    unread_msgs = conv_msg_store[conv_id].copy()
    conv_msg_store[conv_id].clear()
    return jsonify({
        "activities": unread_msgs,
        "watermark": str(len(unread_msgs))
    })

# 供Bot回调发消息的接口
@app.post("/v3/conversations/<conv_id>/activities")
def receive_bot_reply(conv_id):
    activity = request.get_json()
    conv_msg_store[conv_id].append(activity)
    return jsonify({"id": str(uuid.uuid4())})

if __name__ == "__main__":
    app.run(port=PROXY_LISTEN_PORT)

使用步骤 #

• 安装依赖：pip install flask requests
• 先启动3978端口的本地Bot服务，再运行上述代理脚本
• 直接用Postman调用http://localhost:49780下的Direct Line接口即可，和之前通过模拟器开的服务用法完全一致

如果需要测试附件上传、WebSocket流式接收消息、主动推送等高级能力，只需要在上述代码基础上补全对应Direct Line接口的处理逻辑即可，核心的协议转换逻辑不需要改动。

内容的提问来源于stack exchange，提问作者João Pedro Freire Cabral