camel.bots 包

本页内容

camel.bots 包#

子包#

子模块#

camel.bots.telegram_bot 模块#

class camel.bots.telegram_bot.TelegramBot(chat_agent: ChatAgent, telegram_token: str | None = None)[来源]#

基类：object

代表一个由智能体驱动的Telegram机器人。

chat_agent#

为机器人提供动力的聊天代理。

Type:: ChatAgent

telegram_token#

机器人令牌。

Type:: 字符串, 可选

on_message(message: Message) → None[来源]#

处理来自用户的消息。

Parameters:: 消息 (types.Message) – 传入的消息对象。

run() → None[来源]#: 启动Telegram机器人。

模块内容#

基类：object

一个表示使用discord.py库与Discord服务器交互的Discord应用程序的类。

这个机器人可以响应特定频道中的消息，并且只对提及该机器人的消息做出反应。

channel_ids#

允许的频道ID列表。如果提供，机器人将仅在这些频道中响应消息。

Type:: 可选[List[int]]

token#

用于身份验证的Discord机器人令牌。

Type:: 可选[str]

property client#

async exchange_code_for_token_response(code: str) → str | None[来源]#

将授权码交换为访问令牌。

Parameters:

code (str) – 用户授权后从Discord接收的授权码。

Returns:

如果成功则返回访问令牌，否则返回None。

Return type:

可选[str]

Raises:

ValueError - 如果OAuth配置不完整或无效。
httpx.RequestError – 如果在请求过程中出现网络问题。

async get_user_info(access_token: str) → dict | None[来源]#

使用访问令牌检索用户信息。

Parameters:: access_token (str) – 从Discord接收的访问令牌。
Returns:: 从Discord获取的用户信息。
Return type:: 字典

async get_valid_access_token(guild_id: str) → str | None[来源]#

获取指定公会的有效访问令牌。

该方法尝试获取特定公会的访问令牌。如果当前访问令牌已过期，它将使用刷新令牌来更新令牌。

Parameters:

guild_id (str) – 要获取访问令牌的公会ID。

Returns:

如果成功则返回有效的访问令牌，: 否则返回None。

Return type:

可选[str]

async on_message(message: Message) → None[来源]#

用于处理传入消息的事件处理程序。

每当机器人接收到新消息时，都会调用此方法。它会忽略机器人自己发送的消息，只响应允许频道中的消息（如果指定了的话），并且仅响应提及该机器人的消息。

Parameters:: message (discord.Message) – 从Discord接收到的消息对象。

async on_ready() → None[来源]#

当机器人成功连接到Discord服务器时调用的事件处理程序。

当机器人准备就绪并登录到Discord后，它会打印一条显示机器人用户名的消息。

async refresh_access_token(refresh_token: str) → str | None[来源]#

使用刷新令牌来更新访问令牌。

Parameters:: refresh_token (str) – Discord颁发的刷新令牌，可用于获取新的访问令牌。
Returns:: 如果成功则返回新的访问令牌，否则返回None。
Return type:: 可选[str]

async remove_installation(guild: Guild)[来源]#

移除指定公会的安装。

Parameters:: guild (discord.Guild) – 机器人正在被移除的服务器。

run() → None[来源]#

使用Discord机器人的令牌启动它。

该方法启动机器人并使用提供的令牌同步登录到Discord。它会阻塞执行并保持机器人运行。

async save_installation(guild_id: str, access_token: str, refresh_token: str, expires_in: int)[来源]#

保存给定公会的安装信息。

Parameters:

guild_id (str) – 机器人安装所在公会的ID。
access_token (str) - 公会的访问令牌。
refresh_token (str) – 公会的刷新令牌。
expires_in – (int): 访问令牌的过期时间。

async start()[来源]#

使用其令牌异步启动Discord机器人。

该方法启动机器人并异步使用提供的令牌登录Discord。在异步环境中使用时应该等待它。

class camel.bots.SlackApp(token: str | None = None, scopes: str | None = None, signing_secret: str | None = None, client_id: str | None = None, client_secret: str | None = None, redirect_uri_path: str = '/slack/oauth_redirect', installation_store: AsyncInstallationStore | None = None)[来源]#

基类：object

表示一个由Slack Bolt AsyncApp驱动的Slack应用。

该类负责通过设置事件处理器、运行应用服务器以及处理来自Slack的消息和提及等事件，来初始化和管理Slack应用程序。

Parameters:

token (Optional[str]) – 用于身份验证的Slack API令牌。
scopes (可选[str]) – Slack应用的权限范围。
signing_secret (Optional[str]) – 用于验证Slack请求的签名密钥。
client_id (可选[str]) - Slack应用客户端ID。
client_secret (可选[str]) – Slack应用的客户端密钥。
redirect_uri_path (str) – OAuth重定向的URI路径，默认为"/slack/oauth_redirect"。
installation_store (可选[AsyncInstallationStore]) – 用于处理OAuth安装的安装存储。

async app_mention(context: AsyncBoltContext, client: AsyncWebClient, event: Dict[str, Any], body: Dict[str, Any], say: AsyncSay) → None[来源]#

app_mention 事件的事件处理程序。

当有人在Slack中提到该应用时，此方法会被触发。

Parameters:

context (AsyncBoltContext) – 该事件的Slack Bolt上下文。
client (AsyncWebClient) – Slack Web API客户端。
event (Dict[str, Any]) - 应用提及的事件数据。
body (Dict[str, Any]) – 来自Slack的完整请求体。
say (AsyncSay) – 一个用于向频道发送回复的函数。

async handle_request(request: Request) → Response[来源]#

通过请求处理器处理来自Slack的传入请求。

Parameters:: request (Request) – 一个代表传入请求的Starlette请求对象。
Returns:: 由Slack Bolt处理器生成的响应。

mention_me(context: AsyncBoltContext, body: SlackEventBody) → bool[来源]#

检查消息中是否提到了该机器人。

Parameters:

context (AsyncBoltContext) – 该事件的Slack Bolt上下文。
body (SlackEventBody) – Slack事件的主体内容。

Returns:

如果消息中提到该机器人则为True，否则为False。

Return type:

布尔值

async on_message(context: AsyncBoltContext, client: AsyncWebClient, event: Dict[str, Any], body: Dict[str, Any], say: AsyncSay) → None[来源]#

message事件的事件处理程序。

当应用在Slack中收到消息时触发此方法。

Parameters:

context (AsyncBoltContext) – 该事件的Slack Bolt上下文。
client (AsyncWebClient) – Slack Web API客户端。
event (Dict[str, Any]) - 消息的事件数据。
body (Dict[str, Any]) – 来自Slack的完整请求体。
say (AsyncSay) – 一个用于向频道发送回复的函数。

run(port: int = 3000, path: str = '/slack/events', host: str | None = None) → None[来源]#

启动Slack Bolt应用服务器以监听传入的Slack事件。

Parameters:

port (int) – 服务器运行的端口号（默认为3000）。
path (str) – 接收Slack事件的端点路径（默认为"/slack/events"）。
host (可选[str]) – 要绑定服务器的主机名(默认为None)。

setup_handlers() → None[来源]#

为Slack事件设置事件处理程序，例如app_mention和message。

此方法向Slack Bolt应用注册app_mention和on_message事件处理程序，以响应Slack事件。

class camel.bots.SlackAppMentionEventBody(*, token: str, team_id: str, context_team_id: str | None = None, context_enterprise_id: str | None = None, api_app_id: str, event: SlackAppMentionEventProfile, type: str, event_id: str, event_time: int, authorizations: list[SlackAuthProfile] | None = None, is_ext_shared_channel: bool, event_context: str)[来源]#

基类: SlackEventBody

表示Slack事件中提及该应用的整个消息体。

context_team_id: str | None#: 事件的详细资料。对于应用提及(app mentions)则为None。

event: SlackAppMentionEventProfile#: 事件的详细分析

model_config: ClassVar[ConfigDict] = {}#: 模型的配置，应该是一个符合[ConfigDict][pydantic.config.ConfigDict]的字典。

class camel.bots.SlackAppMentionEventProfile(*, user: str, type: str, ts: str, thread_ts: str | None = None, client_msg_id: str, text: str, team: str, blocks: list, channel: str, event_ts: str, channel_type: str | None = None)[来源]#

基类：SlackEventProfile

表示Slack事件的详细配置文件，其中应用程序在消息中被提及。

channel_type: str | None#: Slack频道的类型。对于应用提及的情况，它是None。

model_config: ClassVar[ConfigDict] = {}#: 模型的配置，应该是一个符合[ConfigDict][pydantic.config.ConfigDict]的字典。

class camel.bots.SlackAuthProfile(*, enterprise_id: str | None = None, team_id: str, user_id: str, is_bot: bool, is_enterprise_install: bool)[来源]#

基类：BaseModel

表示Slack事件中的授权配置文件。

事件将包含一个简洁的authorizations字段，显示该事件对其可见的应用程序的一个安装实例。换句话说，authorizations列表将被截断为单个元素。

如果您的应用正在跟踪多个安装方，最好不要依赖授权中列出的单个方作为任何特定的一方。

要获取可以查看事件的完整列表，请在获取应用级令牌后调用apps.event.authorizations.list方法。有关变更的更多信息请阅读此处；这些变更已于2021年2月24日起对现有应用生效。

参考资料：

enterprise_id: str | None#: 与授权关联的企业ID。

is_bot: bool#: 授权用户是否为机器人。

is_enterprise_install: bool#: 授权是否适用于企业安装。

model_config: ClassVar[ConfigDict] = {}#: 模型的配置，应该是一个符合[ConfigDict][pydantic.config.ConfigDict]的字典。

team_id: str#: 与授权关联的团队ID。

user_id: str#: 与该授权关联的用户ID。

class camel.bots.SlackEventBody(*, token: str, team_id: str, context_team_id: str | None, context_enterprise_id: str | None = None, api_app_id: str, event: SlackEventProfile, type: str, event_id: str, event_time: int, authorizations: list[SlackAuthProfile] | None = None, is_ext_shared_channel: bool, event_context: str)[来源]#

基类：BaseModel

表示Slack事件的整个主体，包括事件配置文件、授权和上下文。

api_app_id: str#: 接收事件的Slack应用程序的唯一标识符。

authorizations: list[SlackAuthProfile] | None#: 一个可选的授权列表，用于描述哪些安装可以看到该事件。

context_enterprise_id: str | None#: 共享频道上下文的企业ID（如适用）。

context_team_id: str | None#: 共享频道上下文的团队ID（如果适用）。

event: SlackEventProfile#: 事件的详细分析

event_context: str#: 表示事件上下文的唯一字符串。

event_id: str#: 由Slack分配给此事件的唯一标识符。

event_time: int#: 表示事件触发时间的时间戳（以秒为单位）。

is_ext_shared_channel: bool#: 指示该事件是否属于不同组织之间的共享通道的一部分。

model_config: ClassVar[ConfigDict] = {}#: 模型的配置，应该是一个符合[ConfigDict][pydantic.config.ConfigDict]的字典。

team_id: str#: 事件发生的团队ID。

token: str#: 用于验证事件来源的令牌。

type: str#: 接收到的整体事件类型（例如，‘event_callback’）。

class camel.bots.SlackEventProfile(*, user: str, type: str, ts: str, thread_ts: str | None = None, client_msg_id: str, text: str, team: str, blocks: list, channel: str, event_ts: str, channel_type: str | None)[来源]#

基类：BaseModel

表示Slack事件的详细配置文件，包括用户、消息和上下文数据。

blocks: list#: 消息块列表，提供结构化信息。

channel: str#: 事件发生的Slack频道ID。

channel_type: str | None#: Slack频道的类型（例如'channel'、'im'）。

client_msg_id: str#: 由客户端为消息生成的唯一ID（如果可用）。

event_ts: str#: 事件发生时的特定时间戳。

model_config: ClassVar[ConfigDict] = {}#: 模型的配置，应该是一个符合[ConfigDict][pydantic.config.ConfigDict]的字典。

team: str#: 事件关联团队的ID。

text: str#: 消息内容文本。

thread_ts: str | None#: 线程中父消息的时间戳。

ts: str#: 表示事件触发时间的时间戳。

type: str#: 事件的类型（例如'message'）。

user: str#: 与该事件关联的用户ID。

class camel.bots.TelegramBot(chat_agent: ChatAgent, telegram_token: str | None = None)[来源]#

基类：object

代表一个由智能体驱动的Telegram机器人。

chat_agent#

为机器人提供动力的聊天代理。

Type:: ChatAgent

telegram_token#

机器人令牌。

Type:: 字符串, 可选

on_message(message: Message) → None[来源]#

处理来自用户的消息。

Parameters:: 消息 (types.Message) – 传入的消息对象。

run() → None[来源]#: 启动Telegram机器人。