mirror of
https://github.com/HKUDS/nanobot.git
synced 2026-08-04 08:28:36 +00:00
40 lines
1.1 KiB
Python
40 lines
1.1 KiB
Python
"""Runtime helpers for SDK calls."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping
|
|
from typing import Any
|
|
|
|
|
|
def build_process_direct_kwargs(
|
|
*,
|
|
session_key: str,
|
|
channel: str,
|
|
chat_id: str,
|
|
sender_id: str,
|
|
media: list[str] | None,
|
|
ephemeral: bool,
|
|
attributes: Mapping[str, Any] | None = None,
|
|
on_stream: Any | None = None,
|
|
on_stream_end: Any | None = None,
|
|
) -> dict[str, Any]:
|
|
kwargs: dict[str, Any] = {"session_key": session_key}
|
|
if channel != "cli":
|
|
kwargs["channel"] = channel
|
|
if chat_id != "direct":
|
|
kwargs["chat_id"] = chat_id
|
|
if sender_id != "user":
|
|
kwargs["sender_id"] = sender_id
|
|
if media is not None:
|
|
kwargs["media"] = media
|
|
if ephemeral:
|
|
kwargs["ephemeral"] = True
|
|
kwargs["_run_extra_hooks_for_ephemeral"] = True
|
|
if attributes is not None:
|
|
kwargs["attributes"] = dict(attributes)
|
|
if on_stream is not None:
|
|
kwargs["on_stream"] = on_stream
|
|
if on_stream_end is not None:
|
|
kwargs["on_stream_end"] = on_stream_end
|
|
return kwargs
|