steamship.agents.mixins.transports package#

Submodules#

steamship.agents.mixins.transports.slack module#

class steamship.agents.mixins.transports.slack.SlackBlock(*, type: str | None = None, block_id: str | None = None, elements: List[SlackElement] | None = None)[source]#

Bases: BaseModel

A Slack Block.

block_id: str | None#
elements: List[SlackElement] | None#
to_blocks() → List[Block] | None[source]#

Return self as a list of Steamship Blocks.

type: str | None#

Common types are: - rich_text

class steamship.agents.mixins.transports.slack.SlackContextBehavior(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]#

Bases: Enum

Defines how history between agent and users is tracked.

These specifications are specifically in regard to how the agent interacts with Slack as it pertains to Agent Context.

ENTIRE_CHANNEL = 'entire-channel'#

Agent context is per channel as a whole, which includes bot mentions sent to the top level channel, and across any thread in that channel.

THREADS_ARE_NEW_CONVERSATIONS = 'threads-are-new-conversations'#

Agent context is thread-aware. The top level channel is treated as its own context, and threads have their own contexts.

class steamship.agents.mixins.transports.slack.SlackElement(*, type: str | None = None, elements: List[SlackElement] | None = None, user_id: str | None = None, text: str | None = None)[source]#

Bases: BaseModel

An element of a Slack Block.

elements: List[SlackElement] | None#
text: str | None#
to_blocks() → List[Block] | None[source]#

Return self as a list of Steamship Blocks.

type: str | None#

Common types are: - rich_text_section - user - text

user_id: str | None#
class steamship.agents.mixins.transports.slack.SlackEvent(*, type: str | None = None, user: str | None = None, channel: str | None = None, tab: str | None = None, event_ts: str | None = None, ts: str | None = None, thread_ts: str | None = None, item: str | None = None, client_msg_id: str | None = None, text: str | None = None, blocks: List[SlackBlock] | None = None, team: str | None = None, bot_id: str | None = None)[source]#

Bases: BaseModel

A Slack Event as reported to the Bot.

blocks: List[SlackBlock] | None#
bot_id: str | None#
channel: str | None#
client_msg_id: str | None#
event_ts: str | None#
is_message() → bool[source]#

Return whether this event is a message

item: str | None#
tab: str | None#
team: str | None#
text: str | None#
thread_ts: str | None#
to_blocks() → List[Block] | None[source]#

Return self as a list of Steamship Blocks.

ts: str | None#
type: str | None#
Common types are:

  • app_home_opened

  • app_mention,

  • reaction_added

  • message.channels - A message was posted to a channel

  • message.groups - A message was posted to a private channel

  • message.im - A message was posted to a DM

  • message.mpim - A message was posted to a Group DM

  • message.app_home - A message was posted to the “App Home” chat

  • hello - The client has connected to the server

See more at: https://api.slack.com/events

user: str | None#
class steamship.agents.mixins.transports.slack.SlackRequest(*, token: str | None = None, team_id: str | None = None, api_app_id: str | None = None, event: SlackEvent | None = None, type: str | None = None, event_id: str | None = None, event_time: int | None = None, event_context: str | None = None)[source]#

Bases: BaseModel

The outer object presented when Slack sends a webhook notification.

https://api.slack.com/apis/connections/events-api#event-type-structure

api_app_id: str | None#
event: SlackEvent | None#
event_context: str | None#
event_id: str | None#
event_time: int | None#
team_id: str | None#
token: str | None#
type: str | None#
class steamship.agents.mixins.transports.slack.SlackThreadingBehavior(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]#

Bases: Enum

Defines how responses from the agent will be delivered in response to user mentions.

ALWAYS_THREADED = 'always-threaded'#

Responses from the bot will always be threaded. If the bot was mentioned at the top level of the channel, a new thread will be created for the response.

FOLLOW_THREADS = 'follow-threads'#

If the bot is mentioned from the top-level channel, the response will be in the channel. If the bot is mentioned from within a thread, the response will be to that thread.

class steamship.agents.mixins.transports.slack.SlackTransport(client: Steamship, config: SlackTransportConfig, agent_service: AgentService)[source]#

Bases: Transport

Slack Transport Mixin for Steamship Agent to Slack interaction.

Current support: - Outputs: outputs: text, image, video, audio output - Inputs: text - Message-triggering events: direct mention, bot room entry

The included manifest file below will register a Slack bot which can be added to rooms and communicated with directly. As configured below, it will not receive messages that were not intended for it.

Integration flow:

Slack needs a different style of integration than Telegram, involving a two-way handshake.

The user:

  1. Clicks a “Bot Manifest Link” which is generated by your Steamship Agent Instance This manifest link contains requested access as well as callback URL information.

  2. Clicks “Accept” to create a Slack Bot using that manifest.

  3. Clicks “Install” to install the Slack Bot to a workspace, accepting the permissions

  4. Copy the Oauth Token from the “Settings > Install App” page.

  5. Set the Oauth token for your agent using the set_slack_access_token POST method

At this point, (1) Slack knows about the Agent, and (2) the Agent knows about Slack.

Interacting with the Bot on Slack will trigger a request/response loop in the Agent.

agent_service: AgentService#
bot_token: str | None = None#
build_emit_func(chat_id: str, incoming_message_ts: str, thread_ts: str | None) → Callable[[List[Block], Dict[str, Any]], None][source]#

Return an EmitFun that sends messages to the appropriate Slack channel.

config: SlackTransportConfig#
get_slack_access_token() → str | None[source]#

Return the Slack Access token, which permits the agent to post to Slack.

instance_init()[source]#

Called when the owning AgentService initializes for the first time.

is_slack_token_set() → InvocableResponse[bool][source]#

Return whether the Slack token has been set as a way for a remote UI to check status.

set_slack_access_token(token: str) → InvocableResponse[str][source]#

Set the slack access token.

setting_store_key()[source]#
slack_event(**kwargs) → InvocableResponse[str][source]#

Respond to an inbound event from Slack.

Return the link to install this Agent as a Slack bot.

slack_respond(**kwargs) → InvocableResponse[str][source]#

Respond to an inbound event from Slack.

slack_respond_sync(**kwargs) → InvocableResponse[str][source]#

Respond to inbound Slack events. This is a PUBLIC endpoint.

class steamship.agents.mixins.transports.slack.SlackTransportConfig(*, slackApiBase: str = 'https://slack.com/api/', threadingBehavior: SlackThreadingBehavior = 'follow-threads', contextBehavior: SlackContextBehavior = 'entire-channel')[source]#

Bases: Config

Configuration object for the SlackTransport.

context_behavior: SlackContextBehavior#
slack_api_base: str#
threading_behavior: SlackThreadingBehavior#

steamship.agents.mixins.transports.steamship_widget module#

class steamship.agents.mixins.transports.steamship_widget.SteamshipWidgetTransport(client: Steamship, agent_service: AgentService)[source]#

Bases: Transport

Experimental base class to encapsulate a Steamship web widget communication channel.

answer(**payload) → List[Block][source]#

Endpoint that implements the contract for Steamship embeddable chat widgets. This is a PUBLIC endpoint since these webhooks do not pass a token.

info() → dict[source]#
instance_init()[source]#

The instance init method will be called ONCE by the engine when a new instance of a package or plugin has been created. By default, this does nothing.

message_output: List[Block]#
save_for_emit(blocks: List[Block], metadata: Dict[str, Any])[source]#

steamship.agents.mixins.transports.telegram module#

class steamship.agents.mixins.transports.telegram.TelegramTransport(client: Steamship, config: TelegramTransportConfig, agent_service: AgentService)[source]#

Bases: Transport

Experimental base class to encapsulate a Telegram communication channel.

agent_service: AgentService#
bot_token: str | None = None#
build_emit_func(chat_id: str) → Callable[[List[Block], Dict[str, Any]], None][source]#
config: TelegramTransportConfig#
get_api_root() → str | None[source]#

Return the API root

get_telegram_access_token() → str | None[source]#

Return the Telegram Access token, which permits the agent to post to Telegram.

instance_init()[source]#

The instance init method will be called ONCE by the engine when a new instance of a package or plugin has been created. By default, this does nothing.

is_telegram_token_set() → InvocableResponse[bool][source]#

Return whether the Telegram token has been set as a way for a remote UI to check status.

set_telegram_access_token(token: str) → InvocableResponse[str][source]#

Set the telegram access token.

setting_store_key()[source]#
telegram_bot_info() → dict[source]#
telegram_connect_webhook()[source]#

Register this AgentService with Telegram.

telegram_disconnect_webhook(*args, **kwargs)[source]#

Unsubscribe from Telegram updates.

telegram_respond(**kwargs) → InvocableResponse[str][source]#

Endpoint implementing the Telegram WebHook contract. This is a PUBLIC endpoint since Telegram cannot pass a Bearer token.

telegram_webhook_info() → dict[source]#
class steamship.agents.mixins.transports.telegram.TelegramTransportConfig(*, botToken: str | None = '', apiBase: str | None = 'https://api.telegram.org/bot')[source]#

Bases: Config

api_base: str | None#
bot_token: str | None#

steamship.agents.mixins.transports.transport module#

class steamship.agents.mixins.transports.transport.Transport(client: Steamship)[source]#

Bases: PackageMixin, ABC

client: Steamship#

Base class to encapsulate a communication channel mixin

Intended use is:

class MyBot(PackageService):

def __init__(

self, client: Steamship, config: Dict[str, Any] = None, context: InvocationContext = None

):

super().__init__(client=client, config=config, context=context) self.set_default_agent(TestAgent()) self.add_mixin(

TelegramTransport(client=client, config=self.config, agent_service=self)

)

parse_inbound(payload: dict, context: dict | None = None) → Block | None[source]#
response_for_exception(e: Exception | None, chat_id: str | None = None) → Block[source]#
send(blocks: List[Block], metadata: Dict[str, Any] | None = None)[source]#

Module contents#