channel-sdk

pip install ragbase-channel-sdk

ragbase-cli init channel my-im
cd my-im
vi src/my_im/adapter.py

from ragbase.channel_sdk import Channel, on_inbound, OutboundMessage
 
class MyIMChannel(Channel):
    @on_inbound
    def receive(self, ctx, raw_event):
        # 你家 IM 的消息格式 → 平台标准 inbound message
        return ctx.normalize(
            sender=raw_event["user_id"],
            text=raw_event["body"],
            session_key=raw_event["thread_id"],
        )
 
    def send(self, ctx, msg: OutboundMessage):
        # 平台标准 outbound → 你家 IM 的发送 API
        token = ctx.credentials["im_bot_token"]
        ctx.http.post(
            "https://my-im.internal/send",
            json={"to": msg.recipient, "body": msg.text},
            headers={"Authorization": f"Bearer {token}"},
        )