camel.bots package

camel.bots package#

Subpackages#

Submodules#

camel.bots.telegram_bot module#

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

Bases: object

Represents a Telegram bot that is powered by an agent.

chat_agent#

Chat agent that will power the bot.

Type:: ChatAgent

telegram_token#

The bot token.

Type:: str, optional

on_message(message: Message) → None[source]#

Handles incoming messages from the user.

Parameters:: message (types.Message) – The incoming message object.

run() → None[source]#: Start the Telegram bot.

Module contents#

Bases: object

A class representing a Discord app that uses the discord.py library to interact with Discord servers.

This bot can respond to messages in specific channels and only reacts to messages that mention the bot.

channel_ids#

A list of allowed channel IDs. If provided, the bot will only respond to messages in these channels.

Type:: Optional[List[int]]

token#

The Discord bot token used for authentication.

Type:: Optional[str]

property client#

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

Exchange the authorization code for an access token.

Parameters:

code (str) – The authorization code received from Discord after user authorization.

Returns:

The access token if successful, otherwise None.

Return type:

Optional[str]

Raises:

ValueError – If OAuth configuration is incomplete or invalid.
httpx.RequestError – If there is a network issue during the request.

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

Retrieve user information using the access token.

Parameters:: access_token (str) – The access token received from Discord.
Returns:: The user information retrieved from Discord.
Return type:: dict

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

Retrieve a valid access token for the specified guild.

This method attempts to retrieve an access token for a specific guild. If the current access token is expired, it will refresh the token using the refresh token.

Parameters:

guild_id (str) – The ID of the guild to retrieve the access token for.

Returns:

The valid access token if successful,: otherwise None.

Return type:

Optional[str]

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

Event handler for processing incoming messages.

This method is called whenever a new message is received by the bot. It will ignore messages sent by the bot itself, only respond to messages in allowed channels (if specified), and only to messages that mention the bot.

Parameters:: message (discord.Message) – The message object received from Discord.

async on_ready() → None[source]#

Event handler that is called when the bot has successfully connected to the Discord server.

When the bot is ready and logged into Discord, it prints a message displaying the bot’s username.

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

Refresh the access token using a refresh token.

Parameters:: refresh_token (str) – The refresh token issued by Discord that can be used to obtain a new access token.
Returns:: The new access token if successful, otherwise None.
Return type:: Optional[str]

async remove_installation(guild: Guild)[source]#

Remove the installation for a given guild.

Parameters:: guild (discord.Guild) – The guild from which the bot is being removed.

run() → None[source]#

Start the Discord bot using its token.

This method starts the bot and logs into Discord synchronously using the provided token. It blocks execution and keeps the bot running.

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

Save the installation information for a given guild.

Parameters:

guild_id (str) – The ID of the guild where the bot is installed.
access_token (str) – The access token for the guild.
refresh_token (str) – The refresh token for the guild.
expires_in – (int): The expiration time of the access token.

async start()[source]#

Asynchronously start the Discord bot using its token.

This method starts the bot and logs into Discord asynchronously using the provided token. It should be awaited when used in an async environment.

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)[source]#

Bases: object

Represents a Slack app that is powered by a Slack Bolt AsyncApp.

This class is responsible for initializing and managing the Slack application by setting up event handlers, running the app server, and handling events such as messages and mentions from Slack.

Parameters:

token (Optional[str]) – Slack API token for authentication.
scopes (Optional[str]) – Slack app scopes for permissions.
signing_secret (Optional[str]) – Signing secret for verifying Slack requests.
client_id (Optional[str]) – Slack app client ID.
client_secret (Optional[str]) – Slack app client secret.
redirect_uri_path (str) – The URI path for OAuth redirect, defaults to “/slack/oauth_redirect”.
installation_store (Optional[AsyncInstallationStore]) – The installation store for handling OAuth installations.

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

Event handler for app_mention events.

This method is triggered when someone mentions the app in Slack.

Parameters:

context (AsyncBoltContext) – The Slack Bolt context for the event.
client (AsyncWebClient) – The Slack Web API client.
event (Dict[str, Any]) – The event data for the app mention.
body (Dict[str, Any]) – The full request body from Slack.
say (AsyncSay) – A function to send a response back to the channel.

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

Handles incoming requests from Slack through the request handler.

Parameters:: request (Request) – A Starlette request object representing the incoming request.
Returns:: The response generated by the Slack Bolt handler.

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

Check if the bot is mentioned in the message.

Parameters:

context (AsyncBoltContext) – The Slack Bolt context for the event.
body (SlackEventBody) – The body of the Slack event.

Returns:

True if the bot is mentioned in the message, False otherwise.

Return type:

bool

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

Event handler for message events.

This method is triggered when the app receives a message in Slack.

Parameters:

context (AsyncBoltContext) – The Slack Bolt context for the event.
client (AsyncWebClient) – The Slack Web API client.
event (Dict[str, Any]) – The event data for the message.
body (Dict[str, Any]) – The full request body from Slack.
say (AsyncSay) – A function to send a response back to the channel.

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

Starts the Slack Bolt app server to listen for incoming Slack events.

Parameters:

port (int) – The port on which the server should run (default is 3000).
path (str) – The endpoint path for receiving Slack events (default is “/slack/events”).
host (Optional[str]) – The hostname to bind the server (default is None).

setup_handlers() → None[source]#

Sets up the event handlers for Slack events, such as app_mention and message.

This method registers the app_mention and on_message event handlers with the Slack Bolt app to respond to Slack events.

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)[source]#

Bases: SlackEventBody

Represents the entire body of a Slack event where the app was mentioned in a message.

context_team_id: str | None#: A detailed profile of the event. it’s None for app mentions.

event: SlackAppMentionEventProfile#: A detailed profile of the event

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}#: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'api_app_id': FieldInfo(annotation=str, required=True), 'authorizations': FieldInfo(annotation=Union[list[SlackAuthProfile], NoneType], required=False, default=None), 'context_enterprise_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'context_team_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'event': FieldInfo(annotation=SlackAppMentionEventProfile, required=True), 'event_context': FieldInfo(annotation=str, required=True), 'event_id': FieldInfo(annotation=str, required=True), 'event_time': FieldInfo(annotation=int, required=True), 'is_ext_shared_channel': FieldInfo(annotation=bool, required=True), 'team_id': FieldInfo(annotation=str, required=True), 'token': FieldInfo(annotation=str, required=True), 'type': FieldInfo(annotation=str, required=True)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

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)[source]#

Bases: SlackEventProfile

Represents the detailed profile of a Slack event where the app was mentioned in a message.

channel_type: str | None#: The type of Slack channel. it’s None for app mentions.

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}#: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'blocks': FieldInfo(annotation=list, required=True), 'channel': FieldInfo(annotation=str, required=True), 'channel_type': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'client_msg_id': FieldInfo(annotation=str, required=True), 'event_ts': FieldInfo(annotation=str, required=True), 'team': FieldInfo(annotation=str, required=True), 'text': FieldInfo(annotation=str, required=True), 'thread_ts': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'ts': FieldInfo(annotation=str, required=True), 'type': FieldInfo(annotation=str, required=True), 'user': FieldInfo(annotation=str, required=True)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

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

Bases: BaseModel

Represents the authorization profile within a Slack event.

Events will contain a single, compact authorizations field that shows one installation of your app that the event is visible to. In other words, lists of authorizations will be truncated to one element.

If there’s more than one installing party that your app is keeping track of, it’s best not to rely on the single party listed in authorizations to be any particular one.

To get a full list of who can see events, call the apps.event. authorizations.list method after obtaining an app-level token. Read more on the changes here; they have taken effect for existing apps as of February 24, 2021.

References:

enterprise_id: str | None#: The ID of the enterprise associated with the authorization.

is_bot: bool#: Whether the authorized user is a bot.

is_enterprise_install: bool#: Whether the authorization is for an enterprise installation.

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}#: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'enterprise_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'is_bot': FieldInfo(annotation=bool, required=True), 'is_enterprise_install': FieldInfo(annotation=bool, required=True), 'team_id': FieldInfo(annotation=str, required=True), 'user_id': FieldInfo(annotation=str, required=True)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

team_id: str#: The ID of the team associated with the authorization.

user_id: str#: The ID of the user associated with the authorization.

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)[source]#

Bases: BaseModel

Represents the entire body of a Slack event, including the event profile, authorization, and context.

api_app_id: str#: The unique identifier for the Slack app that received the event.

authorizations: list[SlackAuthProfile] | None#: An optional list of authorizations that describe which installation can see the event.

context_enterprise_id: str | None#: The enterprise ID for the shared channel context, if applicable.

context_team_id: str | None#: The team ID for the shared channel context, if applicable.

event: SlackEventProfile#: A detailed profile of the event

event_context: str#: A unique string representing the context of the event.

event_id: str#: A unique identifier assigned to this event by Slack.

event_time: int#: The timestamp (in seconds) representing when the event was triggered.

is_ext_shared_channel: bool#: Indicates if the event is part of a shared channel between different organizations.

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}#: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'api_app_id': FieldInfo(annotation=str, required=True), 'authorizations': FieldInfo(annotation=Union[list[SlackAuthProfile], NoneType], required=False, default=None), 'context_enterprise_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'context_team_id': FieldInfo(annotation=Union[str, NoneType], required=True), 'event': FieldInfo(annotation=SlackEventProfile, required=True), 'event_context': FieldInfo(annotation=str, required=True), 'event_id': FieldInfo(annotation=str, required=True), 'event_time': FieldInfo(annotation=int, required=True), 'is_ext_shared_channel': FieldInfo(annotation=bool, required=True), 'team_id': FieldInfo(annotation=str, required=True), 'token': FieldInfo(annotation=str, required=True), 'type': FieldInfo(annotation=str, required=True)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

team_id: str#: The ID of the team where the event is happening.

token: str#: The token to verify the source of the event.

type: str#: The overall type of event received (e.g., ‘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)[source]#

Bases: BaseModel

Represents the detailed profile of a Slack event, including user, message, and context data.

blocks: list#: The list of message blocks, providing structured information.

channel: str#: The ID of the Slack channel where the event happened.

channel_type: str | None#: The type of Slack channel (e.g., ‘channel’, ‘im’).

client_msg_id: str#: A unique ID generated by the client for the message (if available).

event_ts: str#: The event-specific timestamp when it occurred.

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}#: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'blocks': FieldInfo(annotation=list, required=True), 'channel': FieldInfo(annotation=str, required=True), 'channel_type': FieldInfo(annotation=Union[str, NoneType], required=True), 'client_msg_id': FieldInfo(annotation=str, required=True), 'event_ts': FieldInfo(annotation=str, required=True), 'team': FieldInfo(annotation=str, required=True), 'text': FieldInfo(annotation=str, required=True), 'thread_ts': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'ts': FieldInfo(annotation=str, required=True), 'type': FieldInfo(annotation=str, required=True), 'user': FieldInfo(annotation=str, required=True)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

team: str#: The ID of the team that the event is associated with.

text: str#: The message content text.

thread_ts: str | None#: The timestamp of the parent message in a thread.

ts: str#: A timestamp representing when the event was triggered.

type: str#: The type of the event (e.g., ‘message’).

user: str#: The ID of the user associated with the event.

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

Bases: object

Represents a Telegram bot that is powered by an agent.

chat_agent#

Chat agent that will power the bot.

Type:: ChatAgent

telegram_token#

The bot token.

Type:: str, optional

on_message(message: Message) → None[source]#

Handles incoming messages from the user.

Parameters:: message (types.Message) – The incoming message object.

run() → None[source]#: Start the Telegram bot.

camel.bots package

On this page

camel.bots package#

Subpackages#

Submodules#

camel.bots.telegram_bot module#

Module contents#