Saltar a contenido

ciel.adapters — Adapters de mensajería

Adapters de mensajería agnósticos al canal (Fase 8). Extienden el contrato de ciel.gateway.adapter (MessagingAdapter / Message) con implementaciones concretas para Microsoft Teams, Discord y Web UI, más un FakeAdapter totalmente offline usado por los tests. Todos los adapters son runtime-agnostic.

ciel.adapters

Canal-agnostic messaging adapters (Fase 8 — madurez de producción).

Este paquete extiende el contrato de ciel.gateway.adapter (MessagingAdapter / Message) con adapters concretos para Microsoft Teams, Discord y Web UI, más un FakeAdapter totalmente offline usado por los tests.

Todos los adapters son runtime-agnostic (no dependen de FastAPI ni de ciel.runtime): send entrega un Message y receive es un async generator que lo emite. El gateway los monta vía routers FastAPI (ciel.gateway) que enqueuan eventos entrantes; los tests usan FakeAdapter para verificar el ciclo sin red.

OFFLINE-SAFE: TeamsAdapter y DiscordAdapter usan httpx (ya dependencia base) para send; si no hay red, send lanza de forma controlada y los routers los alimentan vía webhook (sin red en tests).

__all__ = ['Message', 'MessagingAdapter', 'FakeAdapter', 'TeamsAdapter', 'DiscordAdapter', 'WebUIAdapter'] module-attribute

logger = logging.getLogger(__name__) module-attribute

DiscordAdapter

Bases: MessagingAdapter

Adapter de Discord vía Webhook (send).

send hace POST {"content": ...} a https://discord.com/api/webhooks/<id>/<token>. receive se alimenta vía create_discord_webhook_router que enqueuea eventos MESSAGE_CREATE.

OFFLINE-SAFE: igual que TeamsAdapter, send propaga errores de red de forma controlada; en tests se inyecta un client mock.

Source code in src/ciel/adapters/__init__.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
class DiscordAdapter(MessagingAdapter):
    """Adapter de Discord vía Webhook (``send``).

    ``send`` hace ``POST {"content": ...}`` a
    ``https://discord.com/api/webhooks/<id>/<token>``. ``receive`` se alimenta
    vía ``create_discord_webhook_router`` que enqueuea eventos
    ``MESSAGE_CREATE``.

    OFFLINE-SAFE: igual que TeamsAdapter, ``send`` propaga errores de red de
    forma controlada; en tests se inyecta un ``client`` mock.
    """

    def __init__(
        self,
        webhook_url: Optional[str] = None,
        *,
        channel: Optional[str] = None,
        client: Optional[object] = None,
    ) -> None:
        self.webhook_url = webhook_url
        self.channel = channel
        self._client = client
        self._inbound: "asyncio.Queue[Message]" = asyncio.Queue()

    def _get_client(self):
        if self._client is not None:
            return self._client
        import httpx

        self._client = httpx.AsyncClient(timeout=10.0)
        return self._client

    async def send(self, message: Message) -> None:
        if not self.webhook_url:
            raise ValueError(
                "DiscordAdapter requires webhook_url to send (Discord webhook URL)."
            )
        client = self._get_client()
        payload = {"content": message.content}
        resp = await client.post(self.webhook_url, json=payload)
        status = getattr(resp, "status_code", 200)
        if status >= 400:
            raise RuntimeError(f"Discord webhook returned HTTP {status}")

    async def receive(self) -> AsyncIterator[Message]:
        while True:
            message = await self._inbound.get()
            yield message

    def enqueue(self, message: Message) -> None:
        self._inbound.put_nowait(message)

FakeAdapter

Bases: MessagingAdapter

Adapter de prueba totalmente offline (sin red).

Implementa send y receive sobre colas asyncio.Queue internas. Los tests lo usan para verificar el ciclo send/receive de cualquier consumidor sin tocar la red.

Source code in src/ciel/adapters/__init__.py 
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
64
65
66
67
68
69
70
71
class FakeAdapter(MessagingAdapter):
    """Adapter de prueba totalmente offline (sin red).

    Implementa ``send`` y ``receive`` sobre colas ``asyncio.Queue`` internas.
    Los tests lo usan para verificar el ciclo send/receive de cualquier
    consumidor sin tocar la red.
    """

    def __init__(self) -> None:
        self._inbound: "asyncio.Queue[Message]" = asyncio.Queue()
        self._sent: list[Message] = []
        self._received: list[Message] = []

    async def send(self, message: Message) -> None:
        self._sent.append(message)
        # En un adapter real, send entrega a un canal externo. Aquí lo
        # simulamos poniéndolo también en inbound para permitir bucles de eco.
        await self._inbound.put(message)

    async def receive(self) -> AsyncIterator[Message]:
        while True:
            message = await self._inbound.get()
            self._received.append(message)
            yield message

    # -- helpers de inspección para tests -------------------------------
    def all_sent(self) -> list[Message]:
        return list(self._sent)

    def all_received(self) -> list[Message]:
        return list(self._received)

    async def receive_one(self) -> Message:
        """Drena y devuelve un único mensaje (conveniencia para tests)."""
        message = await self._inbound.get()
        self._received.append(message)
        return message

    def enqueue(self, message: Message) -> None:
        """Empuja un mensaje inbound (usado por los routers de gateway)."""
        self._inbound.put_nowait(message)

enqueue(message: Message) -> None

Empuja un mensaje inbound (usado por los routers de gateway).

Source code in src/ciel/adapters/__init__.py 
69
70
71
def enqueue(self, message: Message) -> None:
    """Empuja un mensaje inbound (usado por los routers de gateway)."""
    self._inbound.put_nowait(message)

receive_one() -> Message async

Drena y devuelve un único mensaje (conveniencia para tests).

Source code in src/ciel/adapters/__init__.py 
63
64
65
66
67
async def receive_one(self) -> Message:
    """Drena y devuelve un único mensaje (conveniencia para tests)."""
    message = await self._inbound.get()
    self._received.append(message)
    return message

Message dataclass

Immutable DTO representing a single inbound or outbound message.

Attributes:

Name Type Description
id str

Unique message identifier. Generated via uuid4 when not supplied by the payload.
channel str

Logical channel the message arrived on (required).
sender Optional[str]

Optional sender identifier (may be None).
content 'str | list[dict[str, Any]]'

Message body / content (required).
metadata Dict[str, Any]

Free-form metadata dict; defaults to an empty dict.
Source code in src/ciel/gateway/adapter.py 
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
@dataclass(frozen=True)
class Message:
    """Immutable DTO representing a single inbound or outbound message.

    Attributes:
        id: Unique message identifier. Generated via ``uuid4`` when not
            supplied by the payload.
        channel: Logical channel the message arrived on (required).
        sender: Optional sender identifier (may be ``None``).
        content: Message body / content (required).
        metadata: Free-form metadata dict; defaults to an empty dict.
    """

    id: str
    channel: str
    sender: Optional[str]
    content: "str | list[dict[str, Any]]"
    metadata: Dict[str, Any] = field(default_factory=dict)

    def text(self) -> str:
        """Extract plain text from content, tolerant to multimodal parts.

        Mirrors :meth:`ciel.runtime.ChatMessage.text`: returns ``str`` content
        verbatim, or concatenates the ``text`` of every ``text`` part in a list.
        """
        content = self.content
        if isinstance(content, str):
            return content
        if not isinstance(content, list):
            return ""
        parts: list[str] = []
        for part in content:
            if isinstance(part, dict) and part.get("type") == "text":
                text = part.get("text", "")
                if isinstance(text, str):
                    parts.append(text)
        return "".join(parts)

text() -> str

Extract plain text from content, tolerant to multimodal parts.

Mirrors :meth:ciel.runtime.ChatMessage.text: returns str content verbatim, or concatenates the text of every text part in a list.

Source code in src/ciel/gateway/adapter.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def text(self) -> str:
    """Extract plain text from content, tolerant to multimodal parts.

    Mirrors :meth:`ciel.runtime.ChatMessage.text`: returns ``str`` content
    verbatim, or concatenates the ``text`` of every ``text`` part in a list.
    """
    content = self.content
    if isinstance(content, str):
        return content
    if not isinstance(content, list):
        return ""
    parts: list[str] = []
    for part in content:
        if isinstance(part, dict) and part.get("type") == "text":
            text = part.get("text", "")
            if isinstance(text, str):
                parts.append(text)
    return "".join(parts)

MessagingAdapter

Abstract base class for messaging adapters.

Subclasses implement :meth:receive (an async generator yielding inbound :class:Message objects) and :meth:send (delivery of an outbound :class:Message).

Source code in src/ciel/gateway/adapter.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
class MessagingAdapter:
    """Abstract base class for messaging adapters.

    Subclasses implement :meth:`receive` (an async generator yielding
    inbound :class:`Message` objects) and :meth:`send` (delivery of an
    outbound :class:`Message`).
    """

    async def receive(self) -> AsyncIterator[Message]:
        """Yield inbound messages as they arrive.

        Implementations should be async generators that block until a
        message is available and then ``yield`` it.
        """
        raise NotImplementedError
        yield  # pragma: no cover  # makes this a generator for type checkers

    async def send(self, message: Message) -> None:
        """Deliver an outbound message.

        Args:
            message: The :class:`Message` to send.
        """
        raise NotImplementedError

receive() -> AsyncIterator[Message] async

Yield inbound messages as they arrive.

Implementations should be async generators that block until a message is available and then yield it.

Source code in src/ciel/gateway/adapter.py 
66
67
68
69
70
71
72
73
async def receive(self) -> AsyncIterator[Message]:
    """Yield inbound messages as they arrive.

    Implementations should be async generators that block until a
    message is available and then ``yield`` it.
    """
    raise NotImplementedError
    yield  # pragma: no cover  # makes this a generator for type checkers

send(message: Message) -> None async

Deliver an outbound message.

Parameters:

Name Type Description Default
message Message

The :class:Message to send.

 required
Source code in src/ciel/gateway/adapter.py 
75
76
77
78
79
80
81
async def send(self, message: Message) -> None:
    """Deliver an outbound message.

    Args:
        message: The :class:`Message` to send.
    """
    raise NotImplementedError

TeamsAdapter

Bases: MessagingAdapter

Adapter de Microsoft Teams vía Incoming Webhook (send).

send hace POST {"text": content} a la webhook_url de un Incoming Webhook de Teams. receive se alimenta vía un router de webhook (create_teams_webhook_router en ciel.gateway) que enqueuea los eventos entrantes.

OFFLINE-SAFE: si no hay red, send propaga el error de httpx de forma controlada; en tests se inyecta un client mock para verificar el payload sin red.

Source code in src/ciel/adapters/__init__.py 
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
class TeamsAdapter(MessagingAdapter):
    """Adapter de Microsoft Teams vía Incoming Webhook (``send``).

    ``send`` hace ``POST {"text": content}`` a la ``webhook_url`` de un
    Incoming Webhook de Teams. ``receive`` se alimenta vía un router de
    webhook (``create_teams_webhook_router`` en ``ciel.gateway``) que enqueuea
    los eventos entrantes.

    OFFLINE-SAFE: si no hay red, ``send`` propaga el error de ``httpx`` de
    forma controlada; en tests se inyecta un ``client`` mock para verificar
    el payload sin red.
    """

    def __init__(
        self,
        webhook_url: Optional[str] = None,
        *,
        channel: Optional[str] = None,
        client: Optional[object] = None,
    ) -> None:
        self.webhook_url = webhook_url
        self.channel = channel
        self._client = client  # inyectable para tests (httpx.AsyncClient mock)
        self._inbound: "asyncio.Queue[Message]" = asyncio.Queue()

    def _get_client(self):
        if self._client is not None:
            return self._client
        import httpx

        self._client = httpx.AsyncClient(timeout=10.0)
        return self._client

    async def send(self, message: Message) -> None:
        if not self.webhook_url:
            raise ValueError(
                "TeamsAdapter requires webhook_url to send (Incoming Webhook URL)."
            )
        client = self._get_client()
        payload = {"text": message.text()}
        # httpx.AsyncClient.post o un mock con la misma firma.
        resp = await client.post(self.webhook_url, json=payload)
        # Si el cliente es un mock sin status_code, no validamos.
        status = getattr(resp, "status_code", 200)
        if status >= 400:
            raise RuntimeError(f"Teams webhook returned HTTP {status}")

    async def receive(self) -> AsyncIterator[Message]:
        while True:
            message = await self._inbound.get()
            yield message

    def enqueue(self, message: Message) -> None:
        self._inbound.put_nowait(message)

WebUIAdapter

Bases: MessagingAdapter

Adapter de Web UI (totalmente offline, sin red).

send escribe a una cola de salida que la UI sondea (polling); receive lee de una cola de entrada alimentada por la UI. Útil para demos locales y para tests sin red.

Source code in src/ciel/adapters/__init__.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
class WebUIAdapter(MessagingAdapter):
    """Adapter de Web UI (totalmente offline, sin red).

    ``send`` escribe a una cola de salida que la UI sondea (polling);
    ``receive`` lee de una cola de entrada alimentada por la UI. Útil para
    demos locales y para tests sin red.
    """

    def __init__(self) -> None:
        self._outbound: "asyncio.Queue[Message]" = asyncio.Queue()
        self._inbound: "asyncio.Queue[Message]" = asyncio.Queue()

    async def send(self, message: Message) -> None:
        # La UI sondea esta cola para mostrar mensajes salientes.
        self._outbound.put_nowait(message)

    async def receive(self) -> AsyncIterator[Message]:
        while True:
            message = await self._inbound.get()
            yield message

    def enqueue(self, message: Message) -> None:
        self._inbound.put_nowait(message)

    async def poll_outbound(self) -> Message:
        """La UI llama esto para obtener el siguiente mensaje saliente."""
        return await self._outbound.get()

poll_outbound() -> Message async

La UI llama esto para obtener el siguiente mensaje saliente.

Source code in src/ciel/adapters/__init__.py 
207
208
209
async def poll_outbound(self) -> Message:
    """La UI llama esto para obtener el siguiente mensaje saliente."""
    return await self._outbound.get()