API 参考

本页列出插件作者和应用代码可直接从 iamai 导入的公共接口。

Public package exports for iamai plugin authors and application code.

exception iamai.AgentError[source]

Bases: RuntimeError

Raised when an agent helper cannot complete a model or tool operation.

class iamai.AgentTrace(name, status='running', events=<factory>)[source]

Bases: object

Append-only trace used to inspect model calls, tool calls, and observations.

Parameters:

  • name (str)

  • status (str)

  • events (list[TraceEvent])

name: str
status: str
events: list[TraceEvent]
add(kind, name, *, input=None, output=None, **metadata)[source]

Append a trace event and return it.

Parameters:

  • kind (str)

  • name (str)

  • input (Any)

  • output (Any)

  • metadata (Any)

Return type:

TraceEvent

lines(*, limit=12)[source]

Return compact human-readable trace lines.

Parameters:

limit (int)

Return type:

list[str]

mark(status)[source]

Set the final trace status once a turn finishes.

Parameters:

status (str)

Return type:

None

to_dict()[source]

Serialize the full trace into a JSON-compatible dictionary.

Return type:

dict[str, Any]

class iamai.Adapter(runtime, config=None)[source]

Bases: ABC

Transport bridge between iamai and an external protocol or runtime.

Parameters:

  • runtime (Runtime)

  • config (dict[str, Any] | None)

async call_api(action, **params)[source]

Call an adapter-specific API action.

Parameters:

  • action (str)

  • params (Any)

Return type:

Any

async close()[source]

Close adapter resources before the runtime shuts down.

Return type:

None

async emit(event)[source]

Emit an event and return whether the runtime admitted it.

Parameters:

event (Event)

Return type:

bool

name = 'adapter'
abstractmethod async send_message(message, *, event=None, target=None)[source]

Send a message to an adapter-specific event or target.

Parameters:

  • message (Message)

  • event (Event | None)

  • target (Any | None)

Return type:

Any

abstractmethod async start()[source]

Start the adapter and begin receiving external events.

Return type:

None

class iamai.AuditLogger(logger_name='iamai.audit')[source]

Bases: object

Structured audit logger that emits one JSON object per runtime event.

Parameters:

logger_name (str)

emit(action, *, outcome='ok', level='INFO', **fields)[source]

Emit one structured audit record.

Parameters:

  • action (str)

  • outcome (str)

  • level (int | str)

  • fields (Any)

Return type:

None

class iamai.Guardrail(*blocked_tokens)[source]

Bases: object

Simple token-based output guardrail.

Parameters:

blocked_tokens (str)

check(text)[source]

Raise AgentError if text contains a blocked token.

Parameters:

text (str)

Return type:

None

class iamai.JsonStateStore(path)[source]

Bases: StateStore

JSON-file backed plugin state store.

Parameters:

path (str | Path)

load_plugin_state(plugin_name)[source]

Load a copy of one plugin’s JSON-backed state.

Parameters:

plugin_name (str)

Return type:

dict[str, Any]

save_plugin_state(plugin_name, state)[source]

Persist one plugin’s state to the JSON store.

Parameters:

  • plugin_name (str)

  • state (dict[str, Any])

Return type:

None

class iamai.LLMClient(config=None)[source]

Bases: object

Small async client for text and JSON chat completions.

Parameters:

config (LLMConfig | dict[str, Any] | None)

async chat_json(messages, *, temperature=None, max_tokens=None, schema=None, trace=None)[source]

Call the chat model and parse a JSON object or array.

Parameters:

  • messages (list[dict[str, str]])

  • temperature (float | None)

  • max_tokens (int | None)

  • schema (type[BaseModel] | None)

  • trace (AgentTrace | None)

Return type:

dict[str, Any] | list[Any] | BaseModel

async chat_text(messages, *, temperature=None, max_tokens=None, trace=None, response_format=None)[source]

Call the configured chat model and return stripped text.

Parameters:

  • messages (list[dict[str, str]])

  • temperature (float | None)

  • max_tokens (int | None)

  • trace (AgentTrace | None)

  • response_format (dict[str, str] | None)

Return type:

str

class iamai.LLMConfig(api_key='', base_url=None, model='', temperature=0.7, max_tokens=800, timeout=60.0, extra_body=None)[source]

Bases: object

Connection and generation settings for an OpenAI-compatible chat model.

Parameters:

  • api_key (str)

  • base_url (str | None)

  • model (str)

  • temperature (float)

  • max_tokens (int)

  • timeout (float)

  • extra_body (dict[str, Any] | None)

api_key: str
base_url: str | None
model: str
temperature: float
max_tokens: int
timeout: float
extra_body: dict[str, Any] | None
classmethod from_mapping(payload=None)[source]

Build configuration from a mapping and environment fallbacks.

Parameters:

payload (dict[str, Any] | None)

Return type:

LLMConfig

class iamai.NullStateStore[source]

Bases: StateStore

No-op state store used for memory-only operation.

class iamai.Permission(_executor, name='permission')[source]

Bases: object

Composable async permission predicate.

Parameters:

  • _executor (Callable[['Runtime', 'Context', dict[Any, Any]], Any])

  • name (str)

name: str
async evaluate(runtime, ctx, cache)[source]

Return whether the current context is allowed to enter a handler.

Parameters:
Return type:

bool

class iamai.Rule(_executor, name='rule')[source]

Bases: object

Composable async rule predicate that may also return match payloads.

Parameters:

  • _executor (Callable[['Runtime', 'Context', dict[Any, Any]], Any])

  • name (str)

name: str
async evaluate(runtime, ctx, cache)[source]

Return whether the rule matches and any extracted match payload.

Parameters:
Return type:

tuple[bool, RulePayload]

named(name)[source]

Return this rule with a display name useful in traces and rulesets.

Parameters:

name (str)

Return type:

Rule

with_payload(**payload)[source]

Merge static payload values into a successful rule result.

Parameters:

payload (Any)

Return type:

Rule

class iamai.RuntimeMetrics[source]

Bases: object

Small in-memory counter store for health checks and operator inspection.

increment(name, value=1, **labels)[source]

Increment a labeled counter by value.

Parameters:

  • name (str)

  • value (int)

  • labels (Any)

Return type:

None

series()[source]

Return all metric series sorted for deterministic output.

Return type:

list[MetricSeries]

snapshot()[source]

Return counters keyed by their formatted metric names.

Return type:

dict[str, int]

class iamai.SessionManager(*, max_backlog_keys=1024, max_backlog_per_key=3, backlog_ttl_seconds=300.0)[source]

Bases: object

Coordinate per-session waiter registration and message consumption.

Parameters:

  • max_backlog_keys (int)

  • max_backlog_per_key (int)

  • backlog_ttl_seconds (float)

cancel(key=None)[source]

Cancel waiters, optionally scoped to a single key.

Parameters:

key (str | None)

Return type:

int

configure(*, max_backlog_keys, max_backlog_per_key, backlog_ttl_seconds)[source]

Apply bounded backlog settings without replacing active waiters.

Parameters:

  • max_backlog_keys (int)

  • max_backlog_per_key (int)

  • backlog_ttl_seconds (float)

Return type:

None

async consume(ctx)[source]

Deliver a context to the first waiter that accepts it.

Parameters:

ctx (Context)

Return type:

bool

list_waiters()[source]

Return diagnostic information about active waiters.

Return type:

list[dict[str, Any]]

session_key(ctx)[source]

Return the stable waiter key for a context.

Parameters:

ctx (Context)

Return type:

str

async wait_for(ctx, *, timeout=60.0, key=None, rule=None)[source]

Wait for a future context in the same session.

Parameters:

  • ctx (Context)

  • timeout (float | None)

  • key (str | None)

  • rule (Callable[['Context'], Any] | None)

Return type:

Context

class iamai.SqliteStateStore(path)[source]

Bases: StateStore

SQLite-backed plugin state store.

Parameters:

path (str | Path)

load_plugin_state(plugin_name)[source]

Load one plugin’s state from SQLite.

Parameters:

plugin_name (str)

Return type:

dict[str, Any]

save_plugin_state(plugin_name, state)[source]

Persist one plugin’s state to SQLite.

Parameters:

  • plugin_name (str)

  • state (dict[str, Any])

Return type:

None

class iamai.StateStore[source]

Bases: object

Abstract state backend for plugin persistence.

load_plugin_state(plugin_name)[source]

Load state for one plugin.

Parameters:

plugin_name (str)

Return type:

dict[str, Any]

save_plugin_state(plugin_name, state)[source]

Persist state for one plugin.

Parameters:

  • plugin_name (str)

  • state (dict[str, Any])

Return type:

None

class iamai.ToolRegistry[source]

Bases: object

Registry for named tools that can be called by agent workflows.

async call(name, tool_input=None, *, trace=None, approved=False, approval_callback=None, **kwargs)[source]

Call a registered tool by name.

Parameters:

  • name (str)

  • tool_input (Any)

  • trace (AgentTrace | None)

  • approved (bool)

  • approval_callback (Callable[[Tool, Any], bool | Awaitable[bool]] | None)

  • kwargs (Any)

Return type:

Any

describe()[source]

Return a text listing of available tools.

Return type:

str

list_tools()[source]

Return tool metadata for prompts or diagnostics.

Return type:

list[dict[str, Any]]

register(name, description, callback, *, permission_name='', input_schema=None, audit_fields=(), requires_approval=False, runtime_capabilities=())[source]

Register or replace a named tool.

Parameters:

  • name (str)

  • description (str)

  • callback (Callable[[...], Any])

  • permission_name (str)

  • input_schema (dict[str, Any] | None)

  • audit_fields (tuple[str, ...] | list[str])

  • requires_approval (bool)

  • runtime_capabilities (tuple[str, ...] | list[str])

Return type:

None

class iamai.Runtime(config, *, base_path=None)[source]

Bases: object

Top-level runtime container that owns adapters, plugins, state, and dispatch.

Parameters:

  • config (dict[str, Any])

  • base_path (Path | None)

add_middleware(callback, *, priority=100, phase='around')[source]

Register runtime middleware outside of a plugin class.

Parameters:

  • callback (Callable[[...], Any])

  • priority (int)

  • phase (str)

Return type:

None

audit(action, *, outcome='ok', level=20, **fields)[source]

Emit one structured runtime audit event.

Parameters:

  • action (str)

  • outcome (str)

  • level (int)

  • fields (Any)

Return type:

None

async bootstrap()[source]

Load plugins and adapters, then run plugin startup hooks.

Return type:

None

command_prefixes()[source]

Return configured command prefixes.

Return type:

tuple[str, …]

count_metric(name, value=1, **labels)[source]

Increment a runtime counter.

Parameters:

  • name (str)

  • value (int)

  • labels (Any)

Return type:

None

async dispatch(event, adapter)[source]

Dispatch one event, returning whether all matched handlers were admitted.

Handler admission is atomic per event. If the complete matched handler set cannot fit within the configured capacity, none of its handlers are scheduled.

Parameters:
Return type:

bool

classmethod from_config_file(path)[source]

Create a runtime from a TOML configuration file.

Parameters:

path (str | Path)

Return type:

Runtime

get_adapter(name)[source]

Return a loaded adapter by name.

Parameters:

name (str)

Return type:

Adapter

get_adapter_config(adapter_name)[source]

Return raw configuration for one adapter.

Parameters:

adapter_name (str)

Return type:

dict[str, Any]

get_plugin(name)[source]

Return a loaded plugin by name.

Parameters:

name (str)

Return type:

Plugin

get_plugin_config(plugin_name)[source]

Return raw configuration for one plugin.

Parameters:

plugin_name (str)

Return type:

dict[str, Any]

get_plugin_schema(plugin_name)[source]

Return a plugin configuration JSON schema, if available.

Parameters:

plugin_name (str)

Return type:

dict[str, Any] | None

health()[source]

Return a compact runtime health payload.

Return type:

dict[str, Any]

iter_handlers()[source]

Return all bound handlers in runtime dispatch order.

This exposes the actual bound callbacks for plugins that need advanced introspection. Prefer list_handlers for diagnostics and management API payloads.

Return type:

tuple[BoundHandler, …]

list_adapters()[source]

Return operator-facing metadata for loaded adapters.

Return type:

list[dict[str, Any]]

list_handlers()[source]

Return JSON-friendly metadata for all registered plugin handlers.

Return type:

list[dict[str, Any]]

list_metrics()[source]

Return runtime metric series as dictionaries.

Return type:

list[dict[str, Any]]

list_plugin_traces()[source]

Return trace payloads exposed by loaded plugins.

Return type:

list[dict[str, Any]]

list_plugins()[source]

Return operator-facing metadata for loaded plugins.

Return type:

list[dict[str, Any]]

list_sessions()[source]

Return active session waiters for diagnostics.

Return type:

list[dict[str, Any]]

load_adapters()[source]

Load adapters from the current configuration.

Return type:

None

load_plugins()[source]

Load plugins from the current configuration.

Return type:

None

register_dependency(name, value, *, annotation=None)[source]

Register a value for name-based and type-based dependency injection.

Parameters:

  • name (str)

  • value (Any)

  • annotation (type[Any] | None)

Return type:

None

async reload_config()[source]

Reload configuration, plugins, state backend, and adapters atomically.

Return type:

None

async reload_plugins()[source]

Reload user plugins while keeping the current adapter set.

Return type:

None

request_config_reload()[source]

Schedule a configuration reload outside the current handler task.

Return type:

None

request_plugin_reload()[source]

Schedule a plugin reload outside the current handler task.

Return type:

None

property runtime_config: dict[str, Any]

Return the validated [runtime] configuration.

async serve()[source]

Run the runtime until stopped or an adapter fails.

Return type:

None

async shutdown()[source]

Stop adapters, cancel handler tasks, and run plugin shutdown hooks.

Return type:

None

async stop()[source]

Request graceful process shutdown.

Return type:

None

superusers()[source]

Return configured superuser IDs as strings.

Return type:

set[str]

exception iamai.ConfigValidationError[source]

Bases: ValueError

Raised when a TOML config file fails structural validation.

class iamai.Context(runtime, adapter, plugin, event, handler, matches=<factory>)[source]

Bases: object

Runtime context passed to handlers and middleware.

Parameters:
runtime: Runtime
adapter: Adapter
plugin: Plugin
event: Event
handler: BoundHandler
matches: dict[str, Any]
property args: str

Return command arguments captured during command matching.

async call_api(action, **params)[source]

Call an API action on the current adapter.

Parameters:

  • action (str)

  • params (Any)

Return type:

Any

property command_name: str | None

Return the matched command name, if any.

property config: dict[str, Any]

Return this plugin’s validated configuration mapping.

async reload_plugins()[source]

Schedule a plugin reload after the current handler completes.

Return type:

None

async reply(message)[source]

Send a reply to the event’s default target.

Parameters:

message (str | Message)

Return type:

Any

async send(message, *, target=None)[source]

Send a message to an explicit adapter target.

Parameters:

  • message (str | Message)

  • target (Any | None)

Return type:

Any

property shared_state: dict[str, Any]

Return the runtime-wide shared state mapping.

property state: dict[str, Any]

Return this plugin’s private state mapping.

property text: str

Return the plain text carried by the current event.

async wait_for_message(*, timeout=60.0, rule=None)[source]

Wait for the next message in the same session.

Parameters:

  • timeout (float | None)

  • rule (Callable[[Context], Any] | None)

Return type:

Context

class iamai.Depends(provider, use_cache=True)[source]

Bases: object

Declare that a parameter should be resolved from a dependency provider.

Parameters:

  • provider (Callable[[...], Any] | Any)

  • use_cache (bool)

provider: Callable[[...], Any] | Any
use_cache: bool
class iamai.Event(id, adapter, platform, type, detail_type=None, sub_type=None, user_id=None, channel_id=None, guild_id=None, self_id=None, message=<factory>, raw=<factory>)[source]

Bases: object

Adapter-agnostic event payload used by the dispatch pipeline.

Parameters:

  • id (str)

  • adapter (str)

  • platform (str)

  • type (str)

  • detail_type (str | None)

  • sub_type (str | None)

  • user_id (str | None)

  • channel_id (str | None)

  • guild_id (str | None)

  • self_id (str | None)

  • message (Message)

  • raw (dict[str, Any])

id: str
adapter: str
platform: str
type: str
detail_type: str | None
sub_type: str | None
user_id: str | None
channel_id: str | None
guild_id: str | None
self_id: str | None
message: Message
raw: dict[str, Any]
classmethod from_dict(payload)[source]

Build an event from a normalized dictionary payload.

Parameters:

payload (dict[str, Any])

Return type:

Event

property text: str

Return the plain text representation of the event message.

to_dict()[source]

Serialize the event into a JSON-compatible dictionary.

Return type:

dict[str, Any]

class iamai.Message(value=None, *, core=None)[source]

Bases: object

Protocol-neutral message builder and container.

Parameters:

  • value (str | Iterable[Mapping[str, Any]] | 'Message' | None)

  • core (CoreMessage | None)

append(kind, **data)[source]

Append a generic segment and return this message.

Parameters:

  • kind (str)

  • data (Any)

Return type:

Message

append_text(text)[source]

Append a text segment and return this message.

Parameters:

text (str)

Return type:

Message

copy()[source]

Return a copy of this message.

Return type:

Message

property core: MagicMock

Return the underlying Rust-backed core message object.

classmethod ensure(value)[source]

Return value as a Message instance.

Parameters:

value (str | 'Message' | Iterable[Mapping[str, Any]])

Return type:

Message

extend(other)[source]

Append all segments from another message-like value.

Parameters:

other (str | 'Message' | Iterable[Mapping[str, Any]])

Return type:

Message

classmethod from_onebot11(payload)[source]

Create a message from a OneBot11 message payload.

Parameters:

payload (Any)

Return type:

Message

plain_text()[source]

Return only the textual content of the message.

Return type:

str

render_text()[source]

Render the message into a debug-friendly text representation.

Return type:

str

property segments: list[dict[str, Any]]

Return message segments as plain dictionaries.

to_onebot11()[source]

Convert the message into OneBot11 segment dictionaries.

Return type:

list[dict[str, Any]]

class iamai.FieldCondition(path, op=FieldOp.EXISTS, value=<object object>, source='event', capture_as=None, flags=0)[source]

Bases: object

One field comparison used by field and match_fields.

Parameters:

  • path (str | Sequence[str])

  • op (FieldOp)

  • value (Any)

  • source (Literal['event', 'raw', 'state', 'shared_state', 'matches', 'context'])

  • capture_as (str | None)

  • flags (int)

path: str | Sequence[str]
op: FieldOp
value: Any
source: Literal['event', 'raw', 'state', 'shared_state', 'matches', 'context']
capture_as: str | None
flags: int
class iamai.FieldOp(*values)[source]

Bases: str, Enum

Supported operators for dotted-path field rules.

EXISTS = 'exists'
EQUALS = 'equals'
NOT_EQUALS = 'not_equals'
IN = 'in'
CONTAINS = 'contains'
STARTSWITH = 'startswith'
ENDSWITH = 'endswith'
REGEX = 'regex'
GT = 'gt'
GE = 'ge'
LT = 'lt'
LE = 'le'
iamai.adapter_in(*adapter_names)[source]

Allow only events emitted by one of the selected adapters.

Parameters:

adapter_names (str)

Return type:

Permission

iamai.adapter_is(*names)[source]

Match events emitted by one of the selected adapters.

Parameters:

names (str)

Return type:

Rule

iamai.allow_all()[source]

Create a permission that always allows the handler to run.

Return type:

Permission

iamai.allow_rule()

Create a rule that always matches.

Return type:

Rule

iamai.all_rules(*values)

Create a rule that matches only when every child rule matches.

Parameters:

values (Any)

Return type:

Rule

iamai.any_rules(*values)

Create a rule that matches when at least one child rule matches.

Parameters:

values (Any)

Return type:

Rule

iamai.channel_in(*channel_ids)[source]

Allow only events from one of the selected channel IDs.

Parameters:

channel_ids (str | int)

Return type:

Permission

class iamai.Plugin(runtime)[source]

Bases: object

Base class for user-defined iamai plugins.

Parameters:

runtime (Runtime)

property config: dict[str, Any]

Return a copy of this plugin’s raw configuration mapping.

config_model: type[Any] | None = None
property config_obj: Any | None

Return the validated plugin configuration object, if configured.

description: str = ''
iter_handlers()[source]

Return handlers declared on this plugin instance.

Return type:

list[BoundHandler]

iter_middlewares()[source]

Return middleware callbacks declared on this plugin instance.

Return type:

list[BoundMiddleware]

load_after: tuple[str, ...] = ()
load_before: tuple[str, ...] = ()
name: str | None = None
optional_requires: tuple[str, ...] = ()
property plugin_name: str

Return the effective plugin name used by config and state.

priority = 100
requires: tuple[str, ...] = ()
async shutdown()[source]

Run plugin shutdown logic before unloading.

Return type:

None

async startup()[source]

Run plugin startup logic after loading.

Return type:

None

state_scope: str = 'memory'
iamai.command(*names, prefixes=None, adapters=None, event_types=None, detail_types=None, rule=None, permission=None, priority=100, block=False)[source]

Decorate a plugin method as a command handler.

Parameters:

  • names (str)

  • prefixes (tuple[str, ...] | list[str] | None)

  • adapters (tuple[str, ...] | list[str] | None)

  • event_types (tuple[str, ...] | list[str] | None)

  • detail_types (tuple[str, ...] | list[str] | None)

  • rule (Rule | Callable[[...], Any] | None)

  • permission (Permission | Callable[[...], Any] | None)

  • priority (int)

  • block (bool)

Return type:

Callable[[HandlerFunc], HandlerFunc]

iamai.channel_id_is(*values)[source]

Match events whose channel_id is in values.

Parameters:

values (str | int)

Return type:

Rule

iamai.contains(*tokens, require_all=False)[source]

Match message text containing any token, or every token if requested.

Parameters:

  • tokens (str)

  • require_all (bool)

Return type:

Rule

iamai.deny_all()[source]

Create a permission that always denies the handler.

Return type:

Permission

iamai.deny_rule()

Create a rule that never matches.

Return type:

Rule

iamai.detail_type_is(*names)[source]

Match events whose normalized detail type is in names.

Parameters:

names (str)

Return type:

Rule

iamai.depends(provider, *, use_cache=True)[source]

Return a dependency marker for a handler or middleware parameter.

Parameters:

  • provider (Callable[[...], Any] | Any)

  • use_cache (bool)

Return type:

Any

iamai.endswith(*suffixes)[source]

Match message text that ends with one of suffixes.

Parameters:

suffixes (str)

Return type:

Rule

iamai.event_handler(*, adapters=None, event_types=None, detail_types=None, rule=None, permission=None, priority=100, block=False)[source]

Decorate a plugin method as a generic event handler.

Parameters:

  • adapters (tuple[str, ...] | list[str] | None)

  • event_types (tuple[str, ...] | list[str] | None)

  • detail_types (tuple[str, ...] | list[str] | None)

  • rule (Rule | Callable[[...], Any] | None)

  • permission (Permission | Callable[[...], Any] | None)

  • priority (int)

  • block (bool)

Return type:

Callable[[HandlerFunc], HandlerFunc]

iamai.event_type_is(*names)[source]

Match events whose normalized event type is in names.

Parameters:

names (str)

Return type:

Rule

iamai.field(path, *, source='event', exists=None, equals=<object object>, not_equals=<object object>, in_=None, contains=<object object>, startswith=None, endswith=None, regex=None, gt=<object object>, ge=<object object>, lt=<object object>, le=<object object>, capture_as=None, flags=0)[source]

Match a dotted path on event/raw/state/shared_state/matches/context.

Multiple comparisons are combined with logical AND. When capture_as is provided, the resolved value is exposed in the rule payload.

Parameters:

  • path (str | Sequence[str])

  • source (Literal['event', 'raw', 'state', 'shared_state', 'matches', 'context'])

  • exists (bool | None)

  • equals (Any)

  • not_equals (Any)

  • in_ (Iterable[Any] | None)

  • contains (Any)

  • startswith (str | tuple[str, ...] | None)

  • endswith (str | tuple[str, ...] | None)

  • regex (str | Pattern[str] | None)

  • gt (Any)

  • ge (Any)

  • lt (Any)

  • le (Any)

  • capture_as (str | None)

  • flags (int)

Return type:

Rule

iamai.fullmatch(pattern, *, flags=0)[source]

Match stripped message text with re.fullmatch and expose groups.

Parameters:

  • pattern (str)

  • flags (int)

Return type:

Rule

iamai.group_in(*group_ids)[source]

Allow only group-like events from one of the selected group IDs.

Parameters:

group_ids (str | int)

Return type:

Permission

iamai.group_message()[source]

Match OneBot-style group message events.

Return type:

Rule

iamai.guild_id_is(*values)[source]

Match events whose guild_id is in values.

Parameters:

values (str | int)

Return type:

Rule

iamai.match_fields(*conditions)[source]

Create a rule that requires every field condition to match.

Parameters:

conditions (FieldCondition)

Return type:

Rule

iamai.middleware(*, priority=100, phase='around')[source]

Decorate a plugin method as middleware for the selected dispatch phase.

Parameters:

  • priority (int)

  • phase (Literal['before', 'around', 'after', 'error'])

Return type:

Callable[[MiddlewareFunc], MiddlewareFunc]

iamai.message_handler(*, adapters=None, event_types=None, detail_types=None, startswith=None, contains=None, regex=None, rule=None, permission=None, priority=100, block=False)[source]

Decorate a plugin method as a message handler.

Parameters:

  • adapters (tuple[str, ...] | list[str] | None)

  • event_types (tuple[str, ...] | list[str] | None)

  • detail_types (tuple[str, ...] | list[str] | None)

  • startswith (tuple[str, ...] | list[str] | None)

  • contains (tuple[str, ...] | list[str] | None)

  • regex (str | None)

  • rule (Rule | Callable[[...], Any] | None)

  • permission (Permission | Callable[[...], Any] | None)

  • priority (int)

  • block (bool)

Return type:

Callable[[HandlerFunc], HandlerFunc]

iamai.none_of(*values)[source]

Create a rule that matches only when none of the child rules match.

Parameters:

values (Any)

Return type:

Rule

iamai.permission(func)[source]

Wrap a callable as a dependency-injected permission predicate.

Parameters:

func (Callable[[...], Any])

Return type:

Permission

iamai.permission_predicate(func)

Alias for permission that reads naturally in decorator arguments.

Parameters:

func (Callable[[...], Any])

Return type:

Permission

iamai.platform_is(*names)[source]

Match events emitted by one of the selected platforms.

Parameters:

names (str)

Return type:

Rule

iamai.private_message()[source]

Match OneBot-style private message events.

Return type:

Rule

iamai.raw_field(path, **kwargs)[source]

Match a dotted path inside event.raw.

Parameters:

  • path (str | Sequence[str])

  • kwargs (Any)

Return type:

Rule

iamai.regex(pattern, *, flags=0)[source]

Match message text with re.search and expose named groups.

Parameters:

  • pattern (str)

  • flags (int)

Return type:

Rule

iamai.rule(func)[source]

Wrap a callable as a dependency-injected event matching rule.

Parameters:

func (Callable[[...], bool | Mapping[str, Any] | tuple[bool, Mapping[str, Any] | None] | None | Awaitable[bool | Mapping[str, Any] | tuple[bool, Mapping[str, Any] | None] | None]])

Return type:

Rule

iamai.rule_predicate(func)

Alias for rule that reads naturally in decorator arguments.

Parameters:

func (Callable[[...], Any])

Return type:

Rule

class iamai.RuleCase(name, rule, priority=100)[source]

Bases: object

One named rule inside a Ruleset.

Parameters:

  • name (str)

  • rule (Rule)

  • priority (int)

name: str
rule: Rule
priority: int
class iamai.RuleMatch(name, priority, payload=<factory>)[source]

Bases: object

Result produced by Ruleset evaluation.

Parameters:

  • name (str)

  • priority (int)

  • payload (dict[str, Any])

name: str
priority: int
payload: dict[str, Any]
class iamai.Ruleset(name, cases=())[source]

Bases: object

A priority-ordered collection of named rules.

This mirrors rule-engine style rulesets without changing iamai handler dispatch. Use it inside a single handler rule when you need explainable, named branches.

Parameters:

  • name (str)

  • cases (tuple[RuleCase, ...])

name: str
cases: tuple[RuleCase, ...]
as_rule(*, first=True, capture_as='ruleset')[source]

Convert this ruleset into a single rule.

When first is true, evaluation stops at the first matching case. Otherwise all matching cases are evaluated and returned.

Parameters:

  • first (bool)

  • capture_as (str)

Return type:

Rule

async evaluate(runtime, ctx, cache, *, first=False)[source]

Evaluate the ruleset directly and return matched cases.

Parameters:
Return type:

list[RuleMatch]

when(name, condition, *, priority=100)[source]

Return a new ruleset with one named case appended.

Parameters:

  • name (str)

  • condition (Any)

  • priority (int)

Return type:

Ruleset

iamai.ruleset(name='ruleset')[source]

Create an empty named ruleset.

Parameters:

name (str)

Return type:

Ruleset

iamai.startswith(*prefixes)[source]

Match message text that starts with one of prefixes.

Parameters:

prefixes (str)

Return type:

Rule

iamai.state_field(path, **kwargs)[source]

Match a dotted path inside the current plugin state.

Parameters:

  • path (str | Sequence[str])

  • kwargs (Any)

Return type:

Rule

iamai.superusers(*user_ids)[source]

Allow configured runtime superusers or the explicit IDs passed here.

Parameters:

user_ids (str | int)

Return type:

Permission

iamai.text_equals(*values, ignore_case=False, strip=True)[source]

Match message text exactly against one of values.

Parameters:

  • values (str)

  • ignore_case (bool)

  • strip (bool)

Return type:

Rule

iamai.user_id_is(*values)[source]

Match events whose user_id is in values.

Parameters:

values (str | int)

Return type:

Rule

iamai.user_in(*user_ids)[source]

Allow only events whose user_id appears in user_ids.

Parameters:

user_ids (str | int)

Return type:

Permission

iamai.when_all(*values)[source]

Alias for all_of inspired by ruleset engines.

Parameters:

values (Any)

Return type:

Rule

iamai.when_any(*values)[source]

Alias for any_of inspired by ruleset engines.

Parameters:

values (Any)

Return type:

Rule

iamai.word_in(*words, ignore_case=True)[source]

Match whole words in message text.

Parameters:

  • words (str)

  • ignore_case (bool)

Return type:

Rule

iamai.on_command(*names, prefixes=None, adapters=None, event_types=None, detail_types=None, rule=None, permission=None, priority=100, block=False)

Decorate a plugin method as a command handler.

Parameters:

  • names (str)

  • prefixes (tuple[str, ...] | list[str] | None)

  • adapters (tuple[str, ...] | list[str] | None)

  • event_types (tuple[str, ...] | list[str] | None)

  • detail_types (tuple[str, ...] | list[str] | None)

  • rule (Rule | Callable[[...], Any] | None)

  • permission (Permission | Callable[[...], Any] | None)

  • priority (int)

  • block (bool)

Return type:

Callable[[HandlerFunc], HandlerFunc]

iamai.on_message(*, adapters=None, event_types=None, detail_types=None, startswith=None, contains=None, regex=None, rule=None, permission=None, priority=100, block=False)

Decorate a plugin method as a message handler.

Parameters:

  • adapters (tuple[str, ...] | list[str] | None)

  • event_types (tuple[str, ...] | list[str] | None)

  • detail_types (tuple[str, ...] | list[str] | None)

  • startswith (tuple[str, ...] | list[str] | None)

  • contains (tuple[str, ...] | list[str] | None)

  • regex (str | None)

  • rule (Rule | Callable[[...], Any] | None)

  • permission (Permission | Callable[[...], Any] | None)

  • priority (int)

  • block (bool)

Return type:

Callable[[HandlerFunc], HandlerFunc]

iamai.on_event(*, adapters=None, event_types=None, detail_types=None, rule=None, permission=None, priority=100, block=False)

Decorate a plugin method as a generic event handler.

Parameters:

  • adapters (tuple[str, ...] | list[str] | None)

  • event_types (tuple[str, ...] | list[str] | None)

  • detail_types (tuple[str, ...] | list[str] | None)

  • rule (Rule | Callable[[...], Any] | None)

  • permission (Permission | Callable[[...], Any] | None)

  • priority (int)

  • block (bool)

Return type:

Callable[[HandlerFunc], HandlerFunc]