Messaging SDK (Python)

1
import grpc
2
from astropods_messaging import AgentMessagingStub
3
4
channel = grpc.insecure_channel("localhost:9090")
5
stub = AgentMessagingStub(channel)

1
import queue, grpc
2
from astropods_messaging import (
3
    AgentMessagingStub, ConversationRequest, AgentResponse, ContentChunk,
4
)
5
6
channel = grpc.insecure_channel("localhost:9090")
7
stub = AgentMessagingStub(channel)
8
9
outbound: queue.Queue[ConversationRequest] = queue.Queue()
10
11
def requests():
12
    while True:
13
        yield outbound.get()
14
15
for resp in stub.ProcessConversation(requests()):
16
    if resp.HasField("incoming_message"):
17
        m = resp.incoming_message
18
        reply = AgentResponse(
19
            conversation_id=m.conversation_id,
20
            content=ContentChunk(
21
                type=ContentChunk.END,
22
                content=f"you said: {m.content}",
23
            ),
24
        )
25
        outbound.put(ConversationRequest(agent_response=reply))

1
def send(payload):
2
    outbound.put(ConversationRequest(agent_response=AgentResponse(
3
        conversation_id=cid,
4
        content=payload,
5
    )))
6
7
send(ContentChunk(type=ContentChunk.START, content=""))
8
for token in stream_llm(prompt):
9
    send(ContentChunk(type=ContentChunk.DELTA, content=token))
10
send(ContentChunk(type=ContentChunk.END, content=""))

1
from astropods_messaging import StatusUpdate
2
3
send_response(AgentResponse(
4
    conversation_id=cid,
5
    status=StatusUpdate(status=StatusUpdate.SEARCHING),
6
))
7
send_response(AgentResponse(
8
    conversation_id=cid,
9
    status=StatusUpdate(
10
        status=StatusUpdate.CUSTOM,
11
        custom_message="Querying the knowledge base…",
12
        emoji=":mag:",
13
    ),
14
))

1
for resp in stub.ProcessConversation(requests()):
2
    if resp.HasField("feedback"):
3
        fb = resp.feedback
4
        if fb.HasField("reaction"):
5
            log_feedback(fb.response_id, fb.reaction.type, fb.reaction.added)
6
        elif fb.HasField("stream_control"):
7
            cancel_generation(fb.conversation_id)

1
from astropods_messaging.astro.messaging.v1.service_pb2 import (
2
    ThreadHistoryRequest,
3
)
4
history = stub.GetThreadHistory(ThreadHistoryRequest(
5
    conversation_id=cid, max_messages=50,
6
))
7
for m in history.messages:
8
    ...  # m.content, m.was_edited, m.is_deleted, m.original_content

1
import time, random, grpc
2
3
def stream_with_retry(stub, requests_factory, max_attempts=20):
4
    delay = 0.5
5
    attempt = 0
6
    while True:
7
        try:
8
            for resp in stub.ProcessConversation(requests_factory()):
9
                yield resp
10
            return  # clean end
11
        except grpc.RpcError as e:
12
            attempt += 1
13
            if attempt >= max_attempts:
14
                raise
15
            code = e.code()
16
            if code not in (
17
                grpc.StatusCode.UNAVAILABLE,
18
                grpc.StatusCode.DEADLINE_EXCEEDED,
19
                grpc.StatusCode.INTERNAL,
20
                grpc.StatusCode.RESOURCE_EXHAUSTED,
21
            ):
22
                raise
23
            time.sleep(min(delay, 30) * (0.5 + random.random() * 0.5))
24
            delay = min(delay * 2, 30)

1
import queue, threading, grpc
2
from astropods_messaging import (
3
    AgentMessagingStub, ConversationRequest, AgentResponse,
4
    ContentChunk, StatusUpdate, SuggestedPrompts,
5
)
6
from astropods_messaging.astro.messaging.v1.message_pb2 import PlatformContext
7
8
channel = grpc.insecure_channel("localhost:9090")
9
stub = AgentMessagingStub(channel)
10
outbound: queue.Queue = queue.Queue()
11
12
def send(resp: AgentResponse):
13
    outbound.put(ConversationRequest(agent_response=resp))
14
15
def handle_slack_message(m):
16
    cid = m.conversation_id
17
    kind = m.platform_context.event_kind
18
19
    if kind == PlatformContext.EVENT_KIND_OBSERVED:
20
        return  # listen-only
21
22
    send(AgentResponse(
23
        conversation_id=cid,
24
        status=StatusUpdate(status=StatusUpdate.THINKING),
25
    ))
26
27
    open_thread = kind == PlatformContext.EVENT_KIND_APP_MENTION
28
    send(AgentResponse(
29
        conversation_id=cid,
30
        content=ContentChunk(type=ContentChunk.START, content=""),
31
    ))
32
33
    for token in stream_llm(m.content):
34
        send(AgentResponse(
35
            conversation_id=cid,
36
            content=ContentChunk(type=ContentChunk.DELTA, content=token),
37
        ))
38
39
    send(AgentResponse(
40
        conversation_id=cid,
41
        content=ContentChunk(type=ContentChunk.END, content=""),
42
    ))
43
44
    send(AgentResponse(conversation_id=cid, prompts=SuggestedPrompts(prompts=[
45
        SuggestedPrompts.Prompt(id="p1", title="Show example",
46
                                message="Show me an example"),
47
        SuggestedPrompts.Prompt(id="p2", title="Go deeper",
48
                                message="Can you explain more?"),
49
    ])))
50
51
def requests():
52
    while True:
53
        yield outbound.get()
54
55
for resp in stub.ProcessConversation(requests()):
56
    payload = resp.WhichOneof("payload")
57
    if payload == "incoming_message" and resp.incoming_message.platform == "slack":
58
        threading.Thread(target=handle_slack_message,
59
                         args=(resp.incoming_message,)).start()
60
    elif payload == "feedback" and resp.feedback.HasField("reaction"):
61
        r = resp.feedback.reaction
62
        record_feedback(resp.feedback.response_id, r.type, r.added)

1
import json
2
from astropods_messaging.astro.messaging.v1.response_pb2 import (
3
    ResponseAttachment, CardAttachment,
4
)
5
6
block_kit = {
7
    "blocks": [
8
        {"type": "header", "text": {"type": "plain_text", "text": "Deploy #4837"}},
9
        {"type": "section", "fields": [
10
            {"type": "mrkdwn", "text": "*Status:*\n:white_check_mark: Green"},
11
            {"type": "mrkdwn", "text": "*Region:*\nus-east-1"},
12
        ]},
13
        {"type": "actions", "elements": [
14
            {"type": "button",
15
             "text": {"type": "plain_text", "text": "View logs"},
16
             "action_id": "view_logs", "value": "deploy-4837"},
17
        ]},
18
    ],
19
}
20
21
send(AgentResponse(
22
    conversation_id=cid,
23
    content=ContentChunk(
24
        type=ContentChunk.END,
25
        content="Deploy status:",
26
        attachments=[ResponseAttachment(
27
            card=CardAttachment(platform_card_json=json.dumps(block_kit)),
28
        )],
29
    ),
30
))

1
import queue, threading, grpc
2
from astropods_messaging import (
3
    AgentMessagingStub, ConversationRequest, AgentResponse,
4
    ContentChunk, StatusUpdate, Transcript,
5
)
6
7
channel = grpc.insecure_channel("localhost:9090")
8
stub = AgentMessagingStub(channel)
9
outbound: queue.Queue = queue.Queue()
10
11
audio_buffers: dict[str, bytearray] = {}
12
audio_configs: dict[str, object] = {}
13
14
def send(resp: AgentResponse):
15
    outbound.put(ConversationRequest(agent_response=resp))
16
17
def handle_web_message(m):
18
    cid = m.conversation_id
19
    send(AgentResponse(
20
        conversation_id=cid,
21
        status=StatusUpdate(status=StatusUpdate.GENERATING),
22
    ))
23
    send(AgentResponse(
24
        conversation_id=cid,
25
        content=ContentChunk(type=ContentChunk.START, content=""),
26
    ))
27
    for token in stream_llm(m.content):
28
        send(AgentResponse(
29
            conversation_id=cid,
30
            content=ContentChunk(type=ContentChunk.DELTA, content=token),
31
        ))
32
    send(AgentResponse(
33
        conversation_id=cid,
34
        content=ContentChunk(type=ContentChunk.END, content=""),
35
    ))
36
37
def requests():
38
    while True:
39
        yield outbound.get()
40
41
for resp in stub.ProcessConversation(requests()):
42
    payload = resp.WhichOneof("payload")
43
44
    if payload == "incoming_message" and resp.incoming_message.platform == "web":
45
        threading.Thread(
46
            target=handle_web_message,
47
            args=(resp.incoming_message,),
48
        ).start()
49
50
    elif payload == "audio_config":
51
        cfg = resp.audio_config
52
        audio_configs[cfg.conversation_id] = cfg
53
        audio_buffers[cfg.conversation_id] = bytearray()
54
55
    elif payload == "audio_chunk":
56
        chunk = resp.audio_chunk
57
        # Single-session example — maintain a per-session mapping in real apps
58
        cid = next(iter(audio_configs))
59
        audio_buffers[cid].extend(chunk.data)
60
        if chunk.done:
61
            cfg = audio_configs.pop(cid)
62
            audio = audio_buffers.pop(cid)
63
            text = run_stt(bytes(audio),
64
                           encoding=cfg.encoding,
65
                           sample_rate=cfg.sample_rate)
66
            send(AgentResponse(
67
                conversation_id=cid,
68
                transcript=Transcript(text=text),
69
            ))
70
            handle_web_message(type("M", (), {
71
                "conversation_id": cid, "content": text,
72
            }))

1
from astropods_messaging.astro.messaging.v1.response_pb2 import MessageOptions
2
3
m = resp.incoming_message
4
open_thread = (
5
    m.platform == "slack"
6
    and m.platform_context.event_kind == PlatformContext.EVENT_KIND_APP_MENTION
7
)
8
send(AgentResponse(
9
    conversation_id=m.conversation_id,
10
    content=ContentChunk(
11
        type=ContentChunk.START,
12
        content="",
13
        options=MessageOptions(create_thread=True) if open_thread else None,
14
    ),
15
))
16
# …rest of the loop is identical for Slack and web

1
# Service
2
AgentMessagingStub
3
ConversationRequest, HealthCheckRequest, HealthCheckResponse
4
5
# Messages
6
Message, User, Attachment
7
8
# Responses
9
AgentResponse, ContentChunk, StatusUpdate, SuggestedPrompts
10
ThreadMetadata, Transcript, ErrorResponse
11
12
# Config
13
AgentConfig, AgentToolConfig
14
15
# Audio
16
AudioStreamConfig, AudioChunk, AudioEncoding
17
18
# Feedback
19
PlatformFeedback, MessageReaction, TextFeedback
20
ButtonClick, PromptSelection, StreamControl, MessageEdit, MessageDelete