Integration APIs¶

Cross-platform bridging, interaction dispatch, agent tools, MCP servers, and CSP adapters.

Bridging¶

class chatom.bridge.IdentityMapper[source]¶

Bases: object

Maps user identities across chat backends.

Uses email as the primary join key. Backends are registered so the mapper can perform fetch_user(email=...) lookups.

Example:

mapper = IdentityMapper()
mapper.register_backend("slack", slack_backend)
mapper.register_backend("symphony", symphony_backend)

# Auto-discover a user on all backends by email
await mapper.link_by_email("alice@company.com")

# Resolve from one backend to another
sym_user = await mapper.resolve(slack_user, target="symphony")

# Manual link
mapper.link(slack_user, symphony_user, by="email")

register_backend(name: str, backend: Any) → None[source]¶

Parameters:

name – Short identifier (e.g. "slack", "symphony").
backend – A chatom BackendBase instance.

property backends: list[str]¶: List of registered backend names.

link(*users: User, by: str = 'email', backends: list[str] | None = None) → LinkedIdentity[source]¶

Manually link User objects as the same person.

Parameters:

*users – Two or more User objects from different backends.
by – The field to use as the join key (currently only "email").
backends – Optional list of backend names corresponding to each user. If not provided, uses user.metadata.get("backend") or the user’s class name as a fallback.

Returns:

The linked identity group.

Raises:

ValueError – If fewer than 2 users or join key is missing.

async link_by_email(email: str) → LinkedIdentity | None[source]¶

Discover and link a user across all registered backends by email.

Queries each registered backend with fetch_user(email=...). If found on 2+ backends, links them into an identity group.

Parameters:: email – The email address to search for.
Returns:: The linked identity group, or None if found on fewer than 2 backends.

async link_all_by_email(emails: list[str]) → list[LinkedIdentity][source]¶

Link multiple users by email in batch.

Parameters:: emails – List of email addresses.
Returns:: List of successfully linked identity groups (found on 2+ backends).

async resolve(user: User | str, *, source: str = '', target: str) → User | None[source]¶

Resolve a user from one backend to another.

Looks up the user’s identity group and returns the cached User for the target backend. If not cached, attempts a backend lookup by email.

Parameters:

user – A User object or user ID string.
source – The backend the user is from (required if user is a string, or if the user isn’t already linked).
target – The backend to resolve to.

Returns:

The User on the target backend, or None if not found.

resolve_id(user: User | str, *, source: str = '', target: str) → str | None[source]¶

Synchronously resolve a user ID on the target backend.

Only uses the local cache — does not make backend calls.

Parameters:

user – A User object or user ID string.
source – The source backend.
target – The target backend.

Returns:

The user ID on the target backend, or None.

get_identity(user: User | str, backend: str = '') → LinkedIdentity | None[source]¶

Get the identity group for a user.

Parameters:

user – A User object or user ID string.
backend – The backend name (required if user is a string).

Returns:

The LinkedIdentity group, or None.

property identities: list[LinkedIdentity]¶: All linked identity groups.

clear() → None[source]¶: Clear all mappings.

class chatom.bridge.MessageBridge(source: Any, dest: Any, *, source_name: str = '', dest_name: str = '', identity_mapper: IdentityMapper | None = None, channels: dict[str, str] | None = None, attribution: bool = True, attribution_format: str = '📨 {name} (via {source}):\n')[source]¶

Bases: object

Forwards messages between two chat backends.

Handles: - Format conversion (markdown ↔ HTML ↔ MessageML) - Mention translation via IdentityMapper - Sender attribution - Attachment forwarding

Example:

bridge = MessageBridge(
    source=slack_backend,
    dest=symphony_backend,
    identity_mapper=mapper,
    channels={"C123": "sym_stream_id"},
)

# Forward a single message
await bridge.forward(message)

# Forward with explicit target channel
await bridge.forward(message, to_channel="sym_stream_id")

async forward(message: Message, *, to_channel: str | None = None, include_attachments: bool = True) → Message | None[source]¶

Forward a message from source to dest.

Resolves the target channel
Converts the message to a FormattedMessage
Translates mentions via the IdentityMapper
Prepends sender attribution (if enabled)
Renders for the destination backend and sends

Parameters:

message – The source message to forward.
to_channel – Explicit target channel ID (overrides channel map).
include_attachments – Whether to include attachments.

Returns:

The sent message on the destination, or None on failure.

async forward_many(messages: list[Message], *, to_channel: str | None = None, include_attachments: bool = True) → list[Message][source]¶

Forward multiple messages in order.

Parameters:

messages – Source messages.
to_channel – Target channel (uses channel map if not provided).
include_attachments – Whether to include attachments.

Returns:

List of sent messages (None entries excluded).

Interaction dispatch¶

Interaction handler registry.

Provides a small, dependency-free pub/sub for dispatching Interaction events to registered callbacks, keyed on action_id. Usable with or without CSP.

Example:

registry = InteractionRegistry()

@registry.on("confirm_button")
async def handle_confirm(event):
    await backend.send_message(event.channel_id, "Confirmed!")

# Drive the registry from a backend stream
async for event in backend.stream_interactions():
    await registry.dispatch(event)

chatom.handlers.InteractionHandler¶

Handlers can be sync or async callables accepting a single Interaction argument. Async handlers are awaited; sync handlers are called directly.

alias of Callable[[Interaction], Any | Awaitable[Any]]

class chatom.handlers.InteractionRegistry[source]¶

Bases: object

Dispatch interactions to handlers keyed by action_id.

Handlers are called in registration order. A single handler can be registered for multiple action IDs by calling register() repeatedly. Use action_id="" (or register_default()) to register a catch-all handler that fires when no specific handler matches.

register(action_id: str, handler: Callable[[Interaction], Any | Awaitable[Any]]) → None[source]¶: Register handler for action_id.

register_default(handler: Callable[[Interaction], Any | Awaitable[Any]]) → None[source]¶: Register a catch-all handler.

unregister(action_id: str, handler: Callable[[Interaction], Any | Awaitable[Any]]) → bool[source]¶: Remove a previously registered handler. Returns True if removed.

clear(action_id: str | None = None) → None[source]¶: Remove all handlers, or just those for action_id.

on(action_id: str) → Callable[[Callable[[Interaction], Any | Awaitable[Any]]], Callable[[Interaction], Any | Awaitable[Any]]][source]¶

Decorator form of register().

Example:

@registry.on("my_button")
def handle(event): ...

property action_ids: list[str]¶: Return all registered action IDs (excluding the wildcard).

handlers_for(action_id: str) → list[Callable[[Interaction], Any | Awaitable[Any]]][source]¶: Return the handler list that dispatch() would call.

async dispatch(event: Interaction) → list[Any][source]¶

Dispatch event to all matching handlers.

Returns the list of handler results, in registration order. Exceptions from individual handlers are logged and do not prevent subsequent handlers from running.

dispatch_sync(event: Interaction) → list[Any][source]¶

Sync wrapper around dispatch().

Runs the async dispatcher on the current event loop if one is running, otherwise in a short-lived loop. Useful for CSP nodes and other sync call sites.

Agents¶

Requires chatom[agent].

Model Context Protocol¶

Requires chatom[mcp].

CSP¶

Requires csp.

class chatom.csp.BackendAdapter(*args, **kwargs)[source]¶

Bases:

Placeholder when csp is not installed.

chatom.csp.message_reader(*args, **kwargs)[source]¶: Placeholder when csp is not installed.

chatom.csp.message_writer(*args, **kwargs)[source]¶: Placeholder when csp is not installed.

chatom.csp.interaction_reader(*args, **kwargs)[source]¶: Placeholder when csp is not installed.