Message a bot or group

These operations require a running control plane, streaming extra, compatible native runtime and dedicated sandbox instance/profile. Set EDITH_TOKEN if required. Both can cause model inference and charges; --run explicitly opts in.

Send one message to the selected profile's Bot Chat:

python examples/bot_message.py --url https://CONTROL_PLANE --instance INSTANCE --profile PROFILE --message "Say hello" --run

The native bot_relay.deliver call blocks for a one-turn reply, printed as JSON. The message may remain in Bot Chat; remove it through the deployment if supported.

Download bot_message.py

"""Send one message to the Bot Chat on an explicitly selected sandbox profile."""

import argparse
import asyncio
import json
import os

from edith_hermes import Hermes


async def send(profile, message):
    return await profile.native.bot_relay.deliver(message=message)


async def main(args):
    if not args.run:
        raise SystemExit("Pass --run to deliver a potentially billable bot message")
    async with Hermes.remote(args.url, token=os.getenv("EDITH_TOKEN")) as hermes:
        profile = hermes.instances(args.instance).profiles(args.profile)
        print(json.dumps(await send(profile, args.message), indent=2))


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--url", required=True)
    parser.add_argument("--instance", required=True)
    parser.add_argument("--profile", required=True)
    parser.add_argument("--message", required=True)
    parser.add_argument("--run", action="store_true")
    asyncio.run(main(parser.parse_args()))

Stage a local test file and send it to an existing sandbox room using the group attachment extension:

python examples/group_attachment.py --url https://CONTROL_PLANE --instance INSTANCE --profile PROFILE --room ROOM --file ./note.txt --message "Read this" --run

The script prints the staged attachment, event ID, and send result. The room message and staged attachment can persist; remove them through your deployment if supported. A timeout can leave a write's outcome uncertain, so inspect the room before retrying.

Download group_attachment.py

"""Stage and send a local file to an explicitly selected group room."""

import argparse
import asyncio
import base64
import json
import os
from pathlib import Path
from uuid import uuid4

from edith_hermes import AsyncGroupAttachments, Hermes


async def send(connection, room_id, file_path, message):
    attachments = AsyncGroupAttachments(connection)
    attachment_id, event_id = str(uuid4()), str(uuid4())
    staged = await attachments.stage(
        room_id,
        attachment_id=attachment_id,
        name=file_path.name,
        content_base64=base64.b64encode(file_path.read_bytes()).decode(),
    )
    result = await attachments.send(
        room_id, event_id=event_id, payload={"text": message}, attachment_ids=[staged["id"]]
    )
    return {"attachment": staged, "event_id": event_id, "result": result}


async def main(args):
    if not args.run:
        raise SystemExit("Pass --run to stage and send an attachment")
    async with (
        Hermes.remote(args.url, token=os.getenv("EDITH_TOKEN")) as hermes,
        hermes.instances(args.instance).profiles(args.profile).connect() as connection,
    ):
        print(json.dumps(await send(connection, args.room, args.file, args.message), indent=2))


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--url", required=True)
    parser.add_argument("--instance", required=True)
    parser.add_argument("--profile", required=True)
    parser.add_argument("--room", required=True)
    parser.add_argument("--file", required=True, type=Path)
    parser.add_argument("--message", required=True)
    parser.add_argument("--run", action="store_true")
    asyncio.run(main(parser.parse_args()))