Aller au contenu

Quickstart Channel

Ajouter le fake mémoire et vérifier un message entrant, sa résolution d'identité et sa réponse, sans service externe.

Prérequis

  • Python 3.13 ;
  • uv.

Ajouter La Capability

Depuis un projet Arclith existant :

Bash
1
2
3
4
uvx --from arclith-cli arclith-cli add-adapter \
  --capability channel \
  --adapter memory \
  --yes

La commande crée config/adapters/bidirectional/memory.yaml.

Exécuter Le Flux

Bash
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
uv run python - <<'PY'
import asyncio

from arclith import (
    ChannelDispatcher,
    ChannelHandlerResult,
    ChannelIdentity,
    ChannelIncomingMessage,
    ChannelMessageHandler,
    ChannelOutgoingMessage,
    MemoryChannel,
    MemoryChannelIdentityResolver,
    ResolvedChannelIdentity,
)


class EchoHandler(ChannelMessageHandler):
    async def handle(self, message, identity):
        return ChannelHandlerResult(
            responses=(
                ChannelOutgoingMessage(
                    channel=message.channel,
                    conversation_id=message.conversation_id,
                    thread_id=message.thread_id,
                    text=f"Reçu pour {identity.user_id}: {message.text}",
                ),
            )
        )


async def main():
    external = ChannelIdentity(
        provider="memory",
        external_user_id="external-42",
    )
    resolver = MemoryChannelIdentityResolver()
    resolver.register(
        external,
        ResolvedChannelIdentity(user_id="user-42", tenant_id="tenant-demo"),
    )
    channel = MemoryChannel()
    dispatcher = ChannelDispatcher(EchoHandler(), resolver, channel, channel)

    message = ChannelIncomingMessage(
        channel="memory",
        provider_event_id="event-1",
        conversation_id="conversation-1",
        sender=external,
        text="bonjour",
    )
    result = await dispatcher.dispatch(message)
    duplicate = await dispatcher.dispatch(message)

    assert result.status == "completed"
    assert result.identity.user_id == "user-42"
    assert result.receipts[0].status == "delivered"
    assert channel.sent_messages[0].text == "Reçu pour user-42: bonjour"
    assert duplicate.status == "duplicate"
    print("channel memory: OK")


asyncio.run(main())
PY

Valider Le Bootstrap

Bash
1
2
3
4
5
6
7
8
uv run python - <<'PY'
from arclith import Arclith, MemoryChannel

app = Arclith("config")
assert app.config.adapters.channel.configured_adapters() == ("memory",)
assert isinstance(app.channel_sender("memory"), MemoryChannel)
print("channel config: OK")
PY

Résultat

Le premier dispatch appelle le handler et enregistre une réponse. Le second retourne duplicate sans rappeler l'application.

Limite

Le fake mémoire est mono-processus et non durable. Pour un endpoint public, injecter un ChannelEventStore atomique partagé et suivre les exigences de la capability Channel.

Suite

Lire la référence Channel, puis raccorder le webhook générique signé ou Slack Events API.