ChatOps#

inbox = PostgresWebhookInbox(
    session_factory=inbox_transaction,
    lease_owner=replica_id,
    lease_seconds=30,
)

chat = ChatOps(
    app,
    name="operations",
    providers=(slack, teams, discord),
    jobs=jobs,
    inbox=inbox,
)

chat = ChatOps(
    app,
    name="operations",
    providers=(slack, teams, discord),
    installations=installations,
    identity=identity_resolver,
    authorizer=authorizer,
    audit=audit,
)


@chat.command(
    "deploy",
    action="Release::deploy",
    resource="Release::environment",
    second_factor=300,
)
async def deploy(environment: Literal["staging", "production"]) -> ChatReply:
    return ChatReply.in_channel(f"Deploying {environment}")

identity_resolver = ExternalIdentityResolver(
    store=external_identities,
    federation=OrganizationFederation(organizations, installation_organizations),
)

chat = ChatOps(
    app,
    name="operations",
    admission=ConcurrencyPolicy(32),
    rate_limit=RateLimitPolicy(limit=20, window=60),
)

@chat.command("investigate", execution="durable")
async def investigate(context: ChatContext) -> None:
    request = context.agent_request
    if request is None:
        raise RuntimeError("durable agent request is unavailable")
    async for event in agent.run(request):
        await context.emit(event)

transcript = TestClient(app).chat(
    "operations",
    provider="slack",
    installation="T1",
    tenant="slack:T1",
    actor="U1",
    conversation="C1",
)
turn = await transcript.command("deploy", environment="staging")
assert turn.reply.content == "Deploying staging"

AgentBackend(*args, **kwargs)

run(request: AgentRequest) -> AsyncIterator[AgentEvent]

AgentEvent(kind: "Literal['progress', 'text', 'completed']", content: str | None = None, percent: int | None = None, id: str | None = None) -> None

completed() -> AgentEvent

progress(content: str, *, percent: int | None = None) -> AgentEvent

text(content: str, *, id: str | None = None) -> AgentEvent

AgentRequest(tenant: str, actor: str, conversation: str, prompt: str, correlation: ChatCorrelation, native: Any = None, principal: Any = None) -> None

ChatAuditEvent(tenant: str, actor: str, action: str, correlation: ChatCorrelation) -> None

ChatAdmissionError(problem: ProblemDetail) -> None

ChatContext(provider: str, installation: str, tenant: str, actor: str, conversation: str, delivery_id: str, native: Any, raw: Any = None, command: str | None = None, action: str | None = None, inputs: Mapping[str, Any] = <factory>, response_url: str | None = None, identity: Identity | None = None, principal: Any = None, external_identity: ExternalIdentityKey | None = None, agent_request: AgentRequest | None = None, emit: Callable[[AgentEvent], Awaitable[None]] | None = None, job_context: Any = None, stream_key: str | None = None) -> None

ChatCorrelation(interaction_id: str | None = None, job_id: str | None = None, trace_id: str | None = None, provider_message_id: str | None = None) -> None

ChatOps(app: Any = None, *, name: str, providers: tuple[Any, ...] = (), path: str = '/_wreath/chat', installations: Any = None, identity: Any = None, jobs: Any = None, inbox: Any = None, outbox: Any = None, authorizer: Any = None, audit: Any = None, admission: Any = None, rate_limit: Any = None, progress: Any = None, conversation_store: Any = None, clock: Callable[[], float] = <built-in function time>) -> None

async accept(interaction: Any) -> Any

action(name: str, *, description: str | None = None, prefix: bool = False) -> Callable[[Callable[..., Any]], Callable[..., Any]]

add(provider: Any) -> Any

agent(name: str, backend: AgentBackend, *, action: str, description: str | None = None, resource: Any = 'conversation', second_factor: float | None = None, streams: Any = None) -> Callable[..., Any]

async cancel(job_id: str, *, reason: str) -> bool

async claim_action(interaction: Any, *, store: Any) -> Any

command(name: str, *, description: str | None = None, action: str | None = None, resource: Any = None, second_factor: float | None = None, execution: "Literal['inline', 'durable']" = 'inline', streams: Any = None, retries: int | None = None) -> Callable[[Callable[..., Any]], Callable[..., Any]]

event(name: str, *, description: str | None = None) -> Callable[[Callable[..., Any]], Callable[..., Any]]

manifest(provider: str, *, base_url: str) -> Any

problem(error: BaseException) -> ProblemDetail

replay_size: int

require_tenant(tenant: str, reference: ChatReference) -> ChatReference

async send(*, tenant: str, destination: Any, content: str, idempotency_key: str) -> Any

ChatProgressCoalescer(*, interval: float) -> None

flush(*, now: float) -> _ProgressDelivery

offer(event: AgentEvent, *, now: float) -> _ProgressDelivery | None

ChatReference(tenant: str, id: str) -> None

ChatReply(content: str | None = None, visibility: "Literal['default', 'ephemeral', 'in_channel']" = 'default', blocks: tuple[Mapping[str, Any], ...] = (), native: Mapping[str, Any] = <factory>, adaptive_card: Mapping[str, Any] | None = None) -> None

card(value: Mapping[str, Any]) -> ChatReply

ephemeral(content: str, *, blocks: list[Mapping[str, Any]] | tuple[Mapping[str, Any], ...] = (), native: Mapping[str, Any] | None = None) -> ChatReply

for_provider(provider: str) -> dict[str, Any]

in_channel(content: str, *, blocks: list[Mapping[str, Any]] | tuple[Mapping[str, Any], ...] = (), native: Mapping[str, Any] | None = None) -> ChatReply

text(content: str) -> ChatReply

ExternalIdentityKey(*, subject: str, provider: str | None = None, installation: str | None = None, issuer: str | None = None, tenant: str | None = None) -> None

ExternalIdentityResolver(*, store: Any, federation: Any = None) -> None

async resolve(key: ExternalIdentityKey) -> PrincipalBinding

IdentityLinkChallenge(url: str, state: str) -> None

IdentityResolutionError(reason: str) -> None

InMemoryChatActionStore(*, max_entries: int = 4096, ttl: float = 900.0, clock: Callable[[], float] = <built-in function monotonic>) -> None

async claim(custom_id: str, *, tenant: str, actor: str) -> _IssuedAction | None

async issue(*, workflow: str, decision: str, tenant: str, actor: str) -> _IssuedAction

size: int

PrincipalBinding(*, identity: Identity, external: ExternalIdentityKey | None = None, key: ExternalIdentityKey | None = None, principal: Any = None, tenant: str | None = None) -> None

key: ExternalIdentityKey

Slack(*, signing_secret: str, app_id: str | None = None, installations: SlackInstallationStore | None = None, http_client: Any = None, api_client: Any = None, response_client: Any = None, clock: Callable[[], float] = <built-in function time>, sleep: Callable[[float], Awaitable[None]] = <function sleep>, max_age: int = 300, max_retries: int = 3, replay_entries: int = 4096, event_scopes: Mapping[str, tuple[str, ...]] | None = None) -> None

async call(installation: SlackInstallation, method: str, payload: Mapping[str, Any], *, idempotent: bool = False, idempotency_key: str | None = None) -> Mapping[str, Any]

manifest(chat: Any, *, base_url: str) -> Mapping[str, Any]

async respond(target: SlackResponseURL, reply: Any) -> Any

response_url(url: str, *, installation: str) -> SlackResponseURL

async send(*, tenant: str, destination: SlackDestination, content: str, idempotency_key: str) -> Mapping[str, Any]

SlackDestination(channel_id: str, tenant: str, installation: SlackInstallation) -> None

SlackInstallation(app_id: str, team_id: str | None, bot_token: str, bot_user_id: str, scopes: frozenset[str], enterprise_id: str | None = None, is_enterprise_install: bool = False) -> None

key: str

SlackInstallationStore(*args, **kwargs)

async fetch(*, enterprise_id: str | None, team_id: str | None, is_enterprise_install: bool) -> SlackInstallation | None

SlackRateLimited(retry_after: float) -> None

SlackResponseURL(url: str, installation: str) -> None

ConnectorRequest(method: str, url: str, headers: Mapping[str, str], json: Mapping[str, Any]) -> None

Teams(*, config: TeamsBotConfig, verifier: Any = None, connector: Any = None, installations: Any = None, token_provider: Callable[[], str | Awaitable[str]] | None = None, clock: Callable[[], float] = <built-in function time>) -> None

external_identity(activity: TeamsActivity) -> Any

manifest(chat: Any, *, base_url: str) -> Mapping[str, Any]

async proactive(installation: TeamsInstallation, reply: Any, *, idempotency_key: str | None = None) -> Any

async reply(incoming: TeamsActivity, reply: Any, *, idempotency_key: str | None = None) -> Any

async send(*, tenant: str, destination: Any, content: str, idempotency_key: str) -> Any

TeamsActivity(id: str, kind: str, channel: str, service_url: str, tenant_id: str, aad_object_id: str | None, installation_key: tuple[str, str], conversation_id: str, conversation_type: str, reply_to_id: str, sender_id: str, sender_name: str | None, recipient_id: str, recipient_name: str | None, text: str | None, name: str | None, action: str | None, value: Mapping[str, Any], members_added: tuple[str, ...], raw: Mapping[str, Any], connector_verified: bool = False) -> None

parse(payload: Mapping[str, Any]) -> TeamsActivity

verified() -> TeamsActivity

TeamsBotConfig(app_id: str, app_secret: str, messaging_endpoint: str, allowed_tenants: frozenset[str], login_issuers: Mapping[str, str] = <factory>, max_token_lifetime: int = 3600) -> None

TeamsConnectorError(*, status: int, retry_after: float | None = None) -> None

TeamsConnectorVerifier(config: TeamsBotConfig, *, fetch: Callable[[str], Awaitable[Mapping[str, Any]]] | None = None, clock: Callable[[], float] = <built-in function time>) -> None

async startup() -> int

async verify(authorization: str | None, activity: Mapping[str, Any]) -> None

TeamsContext(activity: TeamsActivity, external_identity: Any, tenant: str | None = None, identity: Any = None, principal: Any = None, action: str | None = None, inputs: dict[str, Any] = <factory>, method: str = 'POST', path: str = '/_wreath/chat/teams', state: State = <factory>, agent_request: Any = None, emit: Any = None, job_context: Any = None) -> None

actor: str

channel_actor_id: str

command: str | None

conversation: str

delivery_id: str

installation: str

native: Mapping[str, Any]

raw: Mapping[str, Any]

TeamsInstallation(tenant_id: str, installation_id: str, service_url: str, conversation_id: str, bot_id: str) -> None

refreshed_from(activity: TeamsActivity, *, connector_verified: bool) -> TeamsInstallation

TeamsManifest(package_id: str, app_id: str, version: str, name: str, short_description: str, long_description: str, developer_name: str, website_url: str, privacy_url: str, terms_url: str, scopes: tuple[str, ...], commands: tuple[Mapping[str, str], ...] = (), entra_resource: str | None = None) -> None

package(*, color_icon: bytes, outline_icon: bytes) -> bytes

render() -> dict[str, Any]

TeamsRefusal(reason: str, message: str, *, status: int = 401) -> None

Discord(*, application_id: str, public_key: bytes | str | None, bot_token: str | None, client: Any = None, deliver: Callable[[Any], Any] | None = None, verify: Callable[[bytes, bytes, bytes], bool] = <function verify_ed25519>, clock: Callable[[], float] = <built-in function time>, signature_max_age: float = 300.0, rate_limiter: DiscordRateLimiter | None = None) -> None

async accept(chat: Any, interaction: DiscordInteraction, *, body: bytes | None = None, sent_at: float | None = None) -> _AcceptedInteraction

manifest(chat: Any, *, base_url: str) -> DiscordManifest

async send(*, tenant: str, destination: DiscordDestination, content: str, idempotency_key: str) -> Any

validate() -> None

DiscordCommand(name: str, type: int = 1, description: str = '', options: tuple[DiscordOption, ...] = (), integration_types: tuple[int, ...] = (0,), contexts: tuple[int, ...] = (0,), default_member_permissions: str | None = None) -> None

as_discord() -> dict[str, Any]

DiscordComponent(custom_id: str, component_type: int, message_id: str | None, values: tuple[Any, ...] = ()) -> None

DiscordDestination(channel_id: str, tenant: str) -> None

DiscordInstallation(kind: "Literal['guild', 'user']", owner_id: str) -> None

DiscordInteraction(id: str, application_id: str, token: str, kind: str, actor: DiscordActor, installation: DiscordInstallation, native: Mapping[str, Any], command: DiscordInvokedCommand | None = None, component: DiscordComponent | None = None, modal: DiscordModal | None = None) -> None

custom_id: str

parse(payload: Mapping[str, Any], *, allow_unknown: bool = False) -> DiscordInteraction

tenant: str

DiscordInteractionVerifier(public_key: bytes | str, *, verify: Callable[[bytes, bytes, bytes], bool] = <function verify_ed25519>, clock: Callable[[], float] = <built-in function time>, max_age: float | None = None) -> None

verify(*, signature: str, timestamp: str, body: bytes) -> None

DiscordManifest(application_id: str, commands: tuple[DiscordCommand, ...] = (), bot_token: str | None = None, rate_limiter: DiscordRateLimiter | None = None) -> None

async sync(client: Any, *, scope: "Literal['global', 'guild']", guild_id: str | None = None) -> DiscordSyncResult

DiscordMessage(content: str = '', poll: Any = None) -> None

DiscordModal(custom_id: str, values: dict[str, Any]) -> None

DiscordNativeDelivery(raw: Mapping[str, Any]) -> None

DiscordOption(name: str, description: str, type: int, required: bool = False, choices: tuple[Any, ...] = (), options: tuple[DiscordOption, ...] = (), min_value: int | float | None = None, max_value: int | float | None = None, min_length: int | None = None, max_length: int | None = None, autocomplete: bool = False) -> None

as_discord() -> dict[str, Any]

DiscordRateLimiter(*, clock: Callable[[], float] = <built-in function monotonic>, sleep: Callable[[float], Awaitable[None]] = <function sleep>, capacity: int = 4096) -> None

async acquire(route: str, major: str | None) -> None

async observe(route: str, major: str | None, response: Any) -> None

DiscordResponder(*, application_id: str, interaction_id: str, token: str, received_at: float, client: Any, clock: Callable[[], float] = <built-in function monotonic>, acknowledged: bool = False, component: bool = False, rate_limiter: DiscordRateLimiter | None = None) -> None

async defer(*, ephemeral: bool = False) -> dict[str, Any]

async defer_update() -> dict[str, Any]

async edit_followup(message_id: str, **message: Any) -> Any

async edit_original(**message: Any) -> Any

async followup(*, ephemeral: bool = False, **message: Any) -> Any

for_component(**options: Any) -> DiscordResponder

from_installation(installation: DiscordInstallation) -> str

from_interaction(interaction: DiscordInteraction) -> str