/
steamship_widget.py
85 lines (63 loc) · 3.03 KB
/
steamship_widget.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
import uuid
from typing import List, Optional
from steamship import Block, Steamship, SteamshipError
from steamship.agents.mixins.transports.transport import Transport
from steamship.agents.schema import Metadata
from steamship.agents.service.agent_service import AgentService, build_context_appending_emit_func
from steamship.invocable import post
API_BASE = "https://api.telegram.org/bot"
class SteamshipWidgetTransport(Transport):
"""Experimental base class to encapsulate a Steamship web widget communication channel."""
message_output: List[Block]
def __init__(self, client: Steamship, agent_service: AgentService):
super().__init__(client=client)
self.agent_service = agent_service
def instance_init(self):
pass
def _instance_deinit(self, *args, **kwargs):
"""Unsubscribe from updates."""
pass
def _send(self, blocks: [Block], metadata: Metadata):
"""Send a response to the client.
TODO: Since this isn't a push, but rather an API return, we need to figure out how to model this.
"""
pass
@post("info")
def info(self) -> dict:
return {}
def _parse_inbound(self, payload: dict, context: Optional[dict] = None) -> Optional[Block]:
"""Parses an inbound Steamship widget message."""
message_text = payload.get("question")
if message_text is None:
raise SteamshipError(f"No 'question' found in Steamship widget message: {payload}")
chat_id = payload.get("chat_session_id", "default")
message_id = str(uuid.uuid4())
result = Block(text=message_text)
result.set_chat_id(str(chat_id))
result.set_message_id(str(message_id))
return result
@post("answer", public=True)
def answer(self, **payload) -> List[Block]:
"""Endpoint that implements the contract for Steamship embeddable chat widgets. This is a PUBLIC endpoint since these webhooks do not pass a token."""
incoming_message = self.parse_inbound(payload)
with self.agent_service.build_default_context(
context_id=incoming_message.chat_id
) as context:
context.chat_history.append_user_message(
text=incoming_message.text, tags=incoming_message.tags
)
context.emit_funcs = [
self.save_for_emit,
# allow web-app access to blocks on emit (making them public)
build_context_appending_emit_func(context=context, make_blocks_public=True),
]
try:
self.agent_service.run_agent(self.agent_service.get_default_agent(), context)
except Exception as e:
self.message_output = [
self.response_for_exception(e, chat_id=incoming_message.chat_id)
]
# We don't call self.steamship_widget_transport.send because the result is the return value
return self.message_output
def save_for_emit(self, blocks: List[Block], metadata: Metadata):
self.message_output = blocks