Chat and events¶
Use a sandbox instance and profile with a configured provider and model for the live prompt. Install the streaming extra, set EDITH_TOKEN if required, and use the control-plane URL and opaque instance/profile IDs. The inventory command is read-only; it prints instance and session data:
python examples/control_plane_inventory.py --url https://CONTROL_PLANE --instance INSTANCE --profile PROFILE
Download control_plane_inventory.py
"""Read control-plane instances and sessions without creating anything."""
import argparse
import asyncio
import json
import os
from edith_hermes import AsyncEdithHermes
async def inspect(sdk, instance_id, profile_id):
instances = await sdk.control_plane.list_instances()
sessions = await sdk.runtime.list_sessions(instance_id=instance_id, profile_id=profile_id)
return {"instances": instances, "sessions": sessions}
async def main(args):
async with AsyncEdithHermes.remote(base_url=args.url, token=os.getenv("EDITH_TOKEN")) as sdk:
print(json.dumps(await inspect(sdk, args.instance, args.profile), indent=2))
if __name__ == "__main__":
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--url", required=True, help="Control-plane HTTP URL")
parser.add_argument("--instance", required=True)
parser.add_argument("--profile", required=True)
asyncio.run(main(parser.parse_args()))
To create a native session, submit a prompt, and receive frames over the same connection:
python examples/native_events.py --url https://CONTROL_PLANE --instance INSTANCE --profile PROFILE --prompt "Say hello" --run
This may incur model charges and execute tools. It prints the session and up to --limit event frames, waiting at most --timeout seconds for each. After each frame it checks session-scoped replay from the cursor captured before submission for message.complete, then prints the resulting status. Idle timeout, disconnection, replay gaps, or the frame limit leave the turn status uncertain; inspect the session before retrying. Server requests are printed and left unanswered. An interactive application must examine each request and respond on its originating connection according to its method. The session uses close_on_disconnect=False, so exiting the viewer does not request cancellation. The session and transcript can persist; inspect any active work and close the session, then delete the stored transcript through your deployment when finished.
"""Create a native session, submit a prompt and display events on the same connection."""
import argparse
import asyncio
import json
import os
from edith_hermes import Hermes
async def run(connection, prompt, limit, timeout):
session = await connection.call(
"session.create", title="SDK example", close_on_disconnect=False
)
session_id = session["session_id"]
print("session:", json.dumps(session), flush=True)
before = await connection.call("session.events.since", session_id=session_id, last_seen=0)
cursor, epoch = before["latest_seq"], before["epoch"]
await connection.call("prompt.submit", session_id=session_id, text=prompt)
events = connection.events()
for _ in range(limit):
try:
frame = await asyncio.wait_for(anext(events), timeout=timeout)
except (asyncio.TimeoutError, ConnectionError, StopAsyncIteration):
print(
"Event stream idle or disconnected; turn status is uncertain. Inspect session",
session_id,
)
return
print(json.dumps(frame))
# Server requests are displayed but never answered automatically.
try:
replay = await connection.call(
"session.events.since", session_id=session_id, last_seen=cursor
)
except (ConnectionError, TimeoutError):
print("Replay unavailable; turn status is uncertain. Inspect session", session_id)
return
if replay["truncated"] or replay["epoch"] != epoch:
print(
"Event replay changed or was truncated; turn status is uncertain. Inspect session",
session_id,
)
return
cursor = replay["latest_seq"]
for event in replay["events"]:
if event.get("type") == "message.complete":
payload = event.get("payload") or {}
status = (
"failed"
if payload.get("error") or payload.get("status") == "error"
else ("stopped" if payload.get("status") == "interrupted" else "completed")
)
print("turn status:", status)
return
print("Event limit reached; turn status is uncertain. Inspect session", session_id)
async def main(args):
if not args.run:
raise SystemExit("Pass --run to create a session and submit a potentially billable prompt")
async with (
Hermes.remote(args.url, token=os.getenv("EDITH_TOKEN")) as hermes,
hermes.instances(args.instance).profiles(args.profile).connect() as connection,
):
await run(connection, args.prompt, args.limit, args.timeout)
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("--prompt", required=True)
parser.add_argument("--run", action="store_true")
parser.add_argument("--limit", type=int, default=10, help="Maximum event frames")
parser.add_argument("--timeout", type=float, default=30, help="Seconds to wait per frame")
asyncio.run(main(parser.parse_args()))
For native command approvals, first inspect pending requests on the exact native session:
python examples/approval.py --url https://CONTROL_PLANE --instance INSTANCE --profile PROFILE --session SESSION
After reviewing a request and obtaining a decision, pass its exact request ID with --request REQUEST --choice deny --run. Supported example choices are once, session, and deny; a decision can permit a command to run. This operation is distinct from responding to an arbitrary WebSocket server request.
"""Inspect pending command approvals; optionally respond to one exact request."""
import argparse
import asyncio
import json
import os
from edith_hermes import Hermes
async def inspect_or_respond(profile, session_id, request_id=None, choice=None):
if request_id is None:
return await profile.native.approval.pending(session_id=session_id)
return await profile.native.approval.respond(
session_id=session_id, request_id=request_id, choice=choice
)
async def main(args):
if bool(args.request) != bool(args.choice):
raise SystemExit("Supply both --request and --choice to respond")
if args.request and not args.run:
raise SystemExit("Pass --run to send the approval decision")
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 inspect_or_respond(profile, args.session, args.request, args.choice), 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("--session", required=True)
parser.add_argument("--request")
parser.add_argument("--choice", choices=["once", "session", "deny"])
parser.add_argument("--run", action="store_true")
asyncio.run(main(parser.parse_args()))