Agent/agent/tools/builtin/feishu/chat_test.py

import asyncio
import json
import os
import sys

# 将项目根目录添加到 python 路径，确保可以正确导入 agent 包
PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "..", ".."))
if PROJECT_ROOT not in sys.path:
    sys.path.append(PROJECT_ROOT)

from agent.tools.builtin.feishu.chat import (
    feishu_get_contact_list,
    feishu_send_message_to_contact,
    feishu_get_contact_replies,
    feishu_get_chat_history
)

async def feishu_tools():
    print("开始测试飞书工具...\n")

    # # 1. 测试获取联系人列表
    # print("--- 测试: feishu_get_contact_list ---")
    # result_list = await feishu_get_contact_list()
    # print(f"标题: {result_list.title}")
    # print(f"输出: {result_list.output}")
    # print("-" * 30 + "\n")
    #
    # # 2. 测试发送消息 (以 '谭景玉' 为例，请确保 contacts.json 中有此人且信息正确)
    contact_name = "谭景玉"
    # print(f"--- 测试: feishu_send_message_to_contact (对象: {contact_name}) ---")
    #
    # 测试发送纯文本
    text_content = "干活"
    print(f"正在发送文本: {text_content}")
    result_send_text = await feishu_send_message_to_contact(contact_name, text_content)
    print(f"标题: {result_send_text.title}")
    print(f"输出: {result_send_text.output}")
    if result_send_text.error:
        print(f"错误: {result_send_text.error}")

    # 测试发送多模态消息 (文本 + 图片)
    # 注意：这里的图片 URL 需要是一个可访问的地址，或者你可以使用 base64 格式
    # multimodal_content = [
    #     {"type": "text", "text": "这是一条多模态测试消息："},
    #     {"type": "image_url", "image_url": {"url": "https://www.baidu.com/img/flexible/logo/pc/result.png"}}
    # ]
    # print(f"\n正在发送多模态消息...")
    # result_send_multi = await feishu_send_message_to_contact(contact_name, multimodal_content)
    # print(f"标题: {result_send_multi.title}")
    # # print(f"输出: {result_send_multi.output}")
    # if result_send_multi.error:
    #     print(f"错误: {result_send_multi.error}")
    # print("-" * 30 + "\n")

    # # 3. 测试获取回复
    # print(f"--- 测试: feishu_get_contact_replies (对象: {contact_name}) ---")
    # result_replies = await feishu_get_contact_replies(contact_name)
    # print(f"标题: {result_replies.title}")
    # print(f"消息详情: {result_replies.output}")
    # print("-" * 30 + "\n")

    # # 4. 测试获取历史记录
    # print(f"--- 测试: feishu_get_chat_history (对象: {contact_name}) ---")
    # result_history = await feishu_get_chat_history(contact_name, page_size=5, page_token="4cXSlmN7uFAnWWU5yfIGMNvUNrBPLlXZREzLcnvUtOcmK2QFKfwEqfbui_UDsR-y8ne0BkzXABiYTAQASh-n7my_3zQp6o3ERRz0bZ4LB5zMvahf8x7OQoso1rjrMaKM")
    # print(f"标题: {result_history.title}")
    # print(f"历史记录输出: {result_history.output}")
    # print("-" * 30 + "\n")

if __name__ == "__main__":
    # 模拟环境变量 (如果在系统环境变量中已设置，此处可省略)
    os.environ["FEISHU_APP_ID"] = "cli_a90fe317987a9cc9"
    os.environ["FEISHU_APP_SECRET"] = "nn2dWuXTiRA2N6xodbm4g0qz1AfM2ayi"
    
    try:
        asyncio.run(feishu_tools())
    except KeyboardInterrupt:
        pass
    except Exception as e:
        print(f"测试过程中出现异常: {e}")