Saltar a contenido

ciel.gateway — Superficie de gateway

Bloques de construcción públicos del gateway:

  • create_control_app — plano de control FastAPI para el runtime del agente.
  • gateway.mcp — cliente MCP (stdio/HTTP) + host MCP + integración con el runtime.
  • WebhookAdapter / SlackAdapter — adapters de mensajería entrante.
  • create_webhook_router, create_slack_webhook_router, create_teams_webhook_router, create_discord_webhook_router, create_webui_router — montan adapters en una app FastAPI existente.
  • mount_mcp_app — app FastAPI que expone el endpoint MCP host (/mcp).
  • make_app — construye la aplicación FastAPI completa del gateway.

ciel.gateway

Ciel gateway surface.

Exposes the public gateway building blocks:

  • :func:gateway.base.create_control_app — FastAPI control plane for the agent runtime (/health, /info, /v1/agent/run, /v1/tools/..., /v1/board/list).
  • :mod:gateway.mcp — MCP client (stdio/HTTP) + MCP server host + runtime integration, implemented as a native JSON-RPC endpoint.
  • :class:gateway.adapter.WebhookAdapter — generic inbound messaging adapter.
  • :func:create_webhook_router — mounts a messaging adapter into an existing FastAPI app so external channels can feed the agent.
  • :func:mount_mcp_app — builds a FastAPI app exposing the MCP host endpoint (/mcp) backed by a runtime dispatcher, ready to serve over HTTP.

__all__ = ['create_control_app', 'MCPClient', 'MCPServer', 'DefaultAgentRuntimeMCPHost', 'MCPHostToolProvider', 'Message', 'MessagingAdapter', 'WebhookAdapter', 'SlackAdapter', 'create_webhook_router', 'create_slack_webhook_router', 'create_teams_webhook_router', 'create_discord_webhook_router', 'create_webui_router', 'mount_mcp_app', 'make_app'] module-attribute

logger = logging.getLogger(__name__) module-attribute

DefaultAgentRuntimeMCPHost

Host DefaultAgentRuntime behavior through an MCP server boundary.

Source code in src/ciel/gateway/mcp/integration.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
class DefaultAgentRuntimeMCPHost:
    """Host DefaultAgentRuntime behavior through an MCP server boundary."""

    def __init__(
        self,
        runtime: Any,
        *,
        audit_sink: Optional[Any] = None,
        tenant_id: Optional[str] = None,
        agent: str = "default",
    ) -> None:
        self.runtime = runtime
        self.audit_sink = audit_sink or NullAuditSink()
        self.tenant_id = tenant_id
        self.agent = agent
        self.server = MCPServer(audit_sink=self.audit_sink, tenant_id=tenant_id)

    async def _emit(self, event: AuditEvent, *, tenant_id: Optional[str] = None) -> AuditEvent:
        normalized = propagate(event, tenant_id=tenant_id)
        await self.audit_sink.write(normalized)
        return normalized

    async def handle_request(self, payload: Mapping[str, Any]) -> Dict[str, Any]:
        response = await self.server.handle(payload)
        await self._emit(
            AuditEvent(
                event="mcp.host.request",
                agent=self.agent,
                data={"method": payload.get("method"), "request_id": payload.get("id")},
            ),
            tenant_id=self.tenant_id,
        )
        return response

MCPClient

MCP protocol client over an explicit transport tenant-aware runtime.

Source code in src/ciel/gateway/mcp/client.py 
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
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
class MCPClient:
    """MCP protocol client over an explicit transport tenant-aware runtime."""

    def __init__(
        self,
        *,
        transport: MCPTransport,
        tenant_id: Optional[str] = None,
        audit_sink: Optional[Any] = None,
        agent: str = "default",
    ) -> None:
        if audit_sink is None:
            audit_sink = NullAuditSink()
        self.transport = transport
        self.tenant_id = tenant_id
        self.audit_sink = audit_sink
        self.agent = agent
        self.server_version: Optional[str] = None

    def _request_id(self) -> str:
        return str(uuid.uuid4())

    async def _send(self, payload: Dict[str, Any]) -> None:
        await self.transport.send(payload)

    async def _receive(self) -> AsyncIterator[Dict[str, Any]]:
        async for message in self.transport.receive():
            yield message

    async def _emit(self, event: AuditEvent) -> AuditEvent:
        normalized = propagate(event, tenant_id=self.tenant_id)
        await self.audit_sink.write(normalized)
        return normalized

    async def initialize(
        self,
        *,
        client_name: str = "ciel-mcp-client",
        client_version: str = "0.1.0",
    ) -> None:
        payload = {
            "jsonrpc": "2.0",
            "id": self._request_id(),
            "method": "initialize",
            "params": {
                "clientName": client_name,
                "clientVersion": client_version,
                "tenant_id": self.tenant_id,
            },
        }
        await self._send(payload)
        async for message in self._receive():
            result = message.get("result")
            if result and isinstance(result, dict):
                self.server_version = result.get("serverVersion") or result.get("version")
            break
        await self._emit(
            AuditEvent(
                event="mcp.client.initialize",
                agent=self.agent,
                data={"client": client_name, "clientVersion": client_version},
            )
        )

MCPHostToolProvider

Bases: ToolProvider

Expose an MCP server as a Ciel ToolProvider.

Source code in src/ciel/gateway/mcp/integration.py 
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
64
65
66
67
68
69
70
71
class MCPHostToolProvider(ToolProvider):
    """Expose an MCP server as a Ciel ToolProvider."""

    def __init__(self, server: Any, *, default_toolset: str = "default") -> None:
        self.server = server
        self.default_toolset = default_toolset

    async def tool_specs(self, toolset: str) -> Sequence[ToolSpec]:
        payload = await self.server.handle(
            {"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {"toolset": toolset or self.default_toolset}}
        )
        result = payload.get("result") or {}
        specs = []
        for spec in result.get("tools", []):
            specs.append(
                ToolSpec(
                    name=spec.get("name", ""),
                    description=spec.get("description", ""),
                    parameters=spec.get("parameters", {}),
                    strict=spec.get("strict", False),
                    metadata=spec.get("metadata", {}),
                )
            )
        return specs

    async def execute(
        self,
        *,
        tenant_id: Optional[str] = None,
        toolset: str,
        name: str,
        arguments: Dict[str, Any],
        tool_call_id: str,
    ) -> ToolResult:
        payload = await self.server.handle(
            {
                "jsonrpc": "2.0",
                "id": tool_call_id,
                "method": "tools/call",
                "params": {
                    "id": tool_call_id,
                    "tenant_id": tenant_id,
                    "name": name,
                    "arguments": arguments,
                    "toolset": toolset or self.default_toolset,
                },
            }
        )
        result = payload.get("result") or {}
        return ToolResult(
            id=result.get("id", tool_call_id),
            name=name,
            output=result.get("output"),
            error=(result.get("error") or {}).get("message") if isinstance(result.get("error"), dict) else result.get("error"),
            metadata=result.get("metadata", {}),
        )

MCPServer

Minimal MCP-style JSON-RPC endpoint with tenant-aware handlers.

Source code in src/ciel/gateway/mcp/server.py 
 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
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 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
class MCPServer:
    """Minimal MCP-style JSON-RPC endpoint with tenant-aware handlers."""

    def __init__(
        self,
        *,
        provider: Optional[ToolProvider] = None,
        tenant_id: Optional[str] = None,
        audit_sink: Optional[Any] = None,
    ) -> None:
        self.provider = provider
        self.tenant_id = tenant_id
        self.audit_sink = audit_sink or NullAuditSink()
        self.handlers: Dict[str, Any] = {
            "initialize": self.handle_initialize,
            "shutdown": self.handle_shutdown,
            "tools/list": self.handle_tools_list,
            "tools/call": self.handle_tools_call,
        }

    async def handle(self, payload: Mapping[str, Any]) -> Dict[str, Any]:
        method = payload.get("method")
        request_id = payload.get("id")
        if method not in self.handlers:
            return self._error(request_id, -32601, "Method not found")
        try:
            result = await self.handlers[method](payload.get("params", {}))
        except Exception as exc:  # pragma: no cover - defensive path
            logger.exception("MCP handler error")
            return self._error(request_id, -32603, str(exc))
        return {"jsonrpc": "2.0", "id": request_id, "result": result}

    async def handle_initialize(self, params: Dict[str, Any]) -> Dict[str, Any]:
        tenant_id = params.get("tenant_id") or self.tenant_id
        self.tenant_id = tenant_id
        event = AuditEvent(
            event="mcp.server.initialize",
            data={
                "clientName": params.get("clientName"),
                "clientVersion": params.get("clientVersion"),
            },
        )
        normalized = propagate(event, tenant_id=tenant_id)
        await self.audit_sink.write(normalized)
        return {"status": "initialized", "version": "0.1.0", "tenant_id": tenant_id}

    async def handle_shutdown(self, params: Dict[str, Any]) -> Dict[str, Any]:
        event = AuditEvent(event="mcp.server.shutdown")
        normalized = propagate(event, tenant_id=self.tenant_id)
        await self.audit_sink.write(normalized)
        return {"status": "shutdown"}

    async def handle_tools_list(self, params: Dict[str, Any]) -> Dict[str, Any]:
        toolset = params.get("toolset") or "default"
        tenant_id = params.get("tenant_id") or self.tenant_id
        specs: Sequence[ToolSpec] = ()
        if self.provider is not None:
            specs = await self.provider.tool_specs(tenant_id, toolset)
        return {"tools": [self._serialize_spec(spec) for spec in specs]}

    async def handle_tools_call(self, params: Dict[str, Any]) -> Dict[str, Any]:
        if self.provider is None:
            return {"error": {"code": -32000, "message": "no tool provider configured"}}
        toolset = params.get("toolset") or "default"
        tenant_id = params.get("tenant_id") or self.tenant_id
        call_id = params.get("id") or params.get("tool_call_id") or str(uuid.uuid4())
        result = await self.provider.execute(
            tenant_id=tenant_id,
            toolset=toolset,
            name=params["name"],
            arguments=params.get("arguments") or {},
            tool_call_id=call_id,
        )
        response: Dict[str, Any] = {
            "id": result.id,
            "name": result.name,
            "output": result.output,
            "metadata": dict(result.metadata),
        }
        if result.error:
            response["error"] = result.error
        return response

    def _serialize_spec(self, spec: ToolSpec) -> Dict[str, Any]:
        payload: Dict[str, Any] = {
            "name": spec.name,
            "description": spec.description,
            "parameters": dict(spec.parameters),
        }
        if spec.strict:
            payload["strict"] = spec.strict
        if spec.metadata:
            payload["metadata"] = dict(spec.metadata)
        return payload

    @staticmethod
    def _error(request_id: Optional[Any], code: int, message: str) -> Dict[str, Any]:
        return {
            "jsonrpc": "2.0",
            "id": request_id,
            "error": {"code": code, "message": 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

SlackAdapter

Bases: MessagingAdapter

Bidirectional Slack adapter built on slack_sdk.WebClient.

Parameters:

Name Type Description Default
token str

A Slack bot/user token (xoxb-...). Stored and used to construct the :class:slack_sdk.WebClient.

 required
channel Optional[str]

Optional default channel (e.g. "#general" or a channel ID). Used by :meth:send when a :class:Message does not carry an explicit channel in its metadata.

 None
bot_user_id Optional[str]

Optional bot user id (U...). Reserved for inbound event filtering; not required for sending.

 None

The Slack WebClient is created lazily so that merely importing this module (or instantiating the adapter) is safe even when slack_sdk is not installed. The client is built on first use of :meth:send / :meth:receive.

Source code in src/ciel/gateway/adapter_slack.py 
 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
 72
 73
 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
128
129
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
181
182
183
184
185
186
187
class SlackAdapter(MessagingAdapter):
    """Bidirectional Slack adapter built on ``slack_sdk.WebClient``.

    Args:
        token: A Slack bot/user token (``xoxb-...``). Stored and used to
            construct the :class:`slack_sdk.WebClient`.
        channel: Optional default channel (e.g. ``"#general"`` or a channel
            ID). Used by :meth:`send` when a :class:`Message` does not carry
            an explicit channel in its ``metadata``.
        bot_user_id: Optional bot user id (``U...``). Reserved for inbound
            event filtering; not required for sending.

    The Slack WebClient is created lazily so that merely importing this module
    (or instantiating the adapter) is safe even when ``slack_sdk`` is not
    installed. The client is built on first use of :meth:`send` / :meth:`receive`.
    """

    def __init__(
        self,
        token: str,
        *,
        channel: Optional[str] = None,
        bot_user_id: Optional[str] = None,
    ) -> None:
        self.token = token
        self.channel = channel
        self.bot_user_id = bot_user_id
        self._client = None  # type: ignore[var-annotated]
        self._inbound: "asyncio.Queue[Message]" = asyncio.Queue()

    def enqueue(self, message: Message) -> None:
        """Push an inbound :class:`Message` onto the internal queue.

        Used by the Slack webhook router to feed real ``message.changed``
        events into the adapter; consumers can drain them via
        :meth:`receive` (which is overridden below to prefer the queue).
        """
        self._inbound.put_nowait(message)

    # -- internal helpers -------------------------------------------------

    def _get_client(self):
        """Return the lazily-constructed ``slack_sdk.WebClient``.

        Raises:
            RuntimeError: If ``slack_sdk`` is not installed.
        """
        if self._client is None:
            if not _SLACK_SDK_AVAILABLE:  # pragma: no cover - SDK missing path
                raise RuntimeError(
                    "slack_sdk is not installed. Install it with "
                    "`uv pip install slack-sdk` (extra: messaging) to use SlackAdapter."
                )
            from slack_sdk import WebClient

            self._client = WebClient(token=self.token)
        return self._client

    def _resolve_channel(self, message: Message) -> str:
        """Pick the destination channel for a message.

        Resolution order: ``message.channel`` (the canonical DTO field) ->
        ``message.metadata['channel']`` (convenience override) -> ``self.channel``.
        """
        channel = message.channel or message.metadata.get("channel") or self.channel
        if not channel:
            raise ValueError(
                "No channel resolved for Slack send: provide message.channel "
                "or pass channel= to SlackAdapter."
            )
        return channel

    # -- public API -------------------------------------------------------

    async def send(self, message: Message) -> None:
        """Send ``message.content`` to Slack via ``chat.postMessage``.

        The destination channel is resolved from ``message.metadata['channel']``
        falling back to the adapter's default ``channel``. A clear
        :class:`ValueError` is raised when no channel can be resolved.

        Network access is required for this call to succeed against the real
        Slack API; tests pass a mocked ``WebClient`` to verify behaviour
        without touching the network.

        Args:
            message: The :class:`Message` to deliver.
        """
        client = self._get_client()
        channel = self._resolve_channel(message)
        # chat_postMessage is synchronous in slack_sdk; run it in a thread so
        # this stays friendly to the asyncio event loop.
        loop = asyncio.get_running_loop()
        await loop.run_in_executor(
            None,
            lambda: client.chat_postMessage(channel=channel, text=message.text()),
        )
        logger.debug("SlackAdapter.send -> channel=%s len=%d", channel, len(message.text()))

    async def receive(self) -> AsyncIterator[Message]:
        """Yield inbound Slack messages via a simple polling loop.

        This is a lenient, best-effort implementation: it relies on Slack's
        ``conversations.history`` to pull recent messages and yields any that
        are not authored by the bot user. It polls every few seconds.

        The primary, fully-tested contract of this adapter is :meth:`send`.
        The receive loop is provided so the adapter is genuinely bidirectional,
        but it is intentionally simple and skips advanced cursor pagination and
        deduplication. Inbound event handling is better served by mounting the
        Slack webhook router (see :func:`create_slack_webhook_router`) onto the
        FastAPI app, which enqueues real ``message.changed`` events.

        The loop runs forever; the caller is responsible for cancellation.
        """
        client = self._get_client()
        if not self.channel:
            logger.warning(
                "SlackAdapter.receive: no default channel set; polling skipped."
            )
            return
        seen: set[str] = set()
        loop = asyncio.get_running_loop()
        while True:
            try:
                response = await loop.run_in_executor(
                    None,
                    lambda: client.conversations_history(channel=self.channel, limit=20),
                )
                for msg in response.get("messages", []):
                    msg_id = msg.get("ts")
                    user = msg.get("user")
                    if msg_id in seen:
                        continue
                    seen.add(msg_id)
                    if self.bot_user_id and user == self.bot_user_id:
                        continue
                    yield Message(
                        id=msg_id,
                        channel=self.channel,
                        sender=user,
                        content=msg.get("text", ""),
                        metadata={"slack": msg},
                    )
            except Exception as exc:  # pragma: no cover - network resilience
                logger.warning("SlackAdapter.receive polling error: %s", exc)
            await asyncio.sleep(5)

enqueue(message: Message) -> None

Push an inbound :class:Message onto the internal queue.

Used by the Slack webhook router to feed real message.changed events into the adapter; consumers can drain them via :meth:receive (which is overridden below to prefer the queue).

Source code in src/ciel/gateway/adapter_slack.py 
71
72
73
74
75
76
77
78
def enqueue(self, message: Message) -> None:
    """Push an inbound :class:`Message` onto the internal queue.

    Used by the Slack webhook router to feed real ``message.changed``
    events into the adapter; consumers can drain them via
    :meth:`receive` (which is overridden below to prefer the queue).
    """
    self._inbound.put_nowait(message)

receive() -> AsyncIterator[Message] async

Yield inbound Slack messages via a simple polling loop.

This is a lenient, best-effort implementation: it relies on Slack's conversations.history to pull recent messages and yields any that are not authored by the bot user. It polls every few seconds.

The primary, fully-tested contract of this adapter is :meth:send. The receive loop is provided so the adapter is genuinely bidirectional, but it is intentionally simple and skips advanced cursor pagination and deduplication. Inbound event handling is better served by mounting the Slack webhook router (see :func:create_slack_webhook_router) onto the FastAPI app, which enqueues real message.changed events.

The loop runs forever; the caller is responsible for cancellation.

Source code in src/ciel/gateway/adapter_slack.py 
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
181
182
183
184
185
186
187
async def receive(self) -> AsyncIterator[Message]:
    """Yield inbound Slack messages via a simple polling loop.

    This is a lenient, best-effort implementation: it relies on Slack's
    ``conversations.history`` to pull recent messages and yields any that
    are not authored by the bot user. It polls every few seconds.

    The primary, fully-tested contract of this adapter is :meth:`send`.
    The receive loop is provided so the adapter is genuinely bidirectional,
    but it is intentionally simple and skips advanced cursor pagination and
    deduplication. Inbound event handling is better served by mounting the
    Slack webhook router (see :func:`create_slack_webhook_router`) onto the
    FastAPI app, which enqueues real ``message.changed`` events.

    The loop runs forever; the caller is responsible for cancellation.
    """
    client = self._get_client()
    if not self.channel:
        logger.warning(
            "SlackAdapter.receive: no default channel set; polling skipped."
        )
        return
    seen: set[str] = set()
    loop = asyncio.get_running_loop()
    while True:
        try:
            response = await loop.run_in_executor(
                None,
                lambda: client.conversations_history(channel=self.channel, limit=20),
            )
            for msg in response.get("messages", []):
                msg_id = msg.get("ts")
                user = msg.get("user")
                if msg_id in seen:
                    continue
                seen.add(msg_id)
                if self.bot_user_id and user == self.bot_user_id:
                    continue
                yield Message(
                    id=msg_id,
                    channel=self.channel,
                    sender=user,
                    content=msg.get("text", ""),
                    metadata={"slack": msg},
                )
        except Exception as exc:  # pragma: no cover - network resilience
            logger.warning("SlackAdapter.receive polling error: %s", exc)
        await asyncio.sleep(5)

send(message: Message) -> None async

Send message.content to Slack via chat.postMessage.

The destination channel is resolved from message.metadata['channel'] falling back to the adapter's default channel. A clear :class:ValueError is raised when no channel can be resolved.

Network access is required for this call to succeed against the real Slack API; tests pass a mocked WebClient to verify behaviour without touching the network.

Parameters:

Name Type Description Default
message Message

The :class:Message to deliver.

 required
Source code in src/ciel/gateway/adapter_slack.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
async def send(self, message: Message) -> None:
    """Send ``message.content`` to Slack via ``chat.postMessage``.

    The destination channel is resolved from ``message.metadata['channel']``
    falling back to the adapter's default ``channel``. A clear
    :class:`ValueError` is raised when no channel can be resolved.

    Network access is required for this call to succeed against the real
    Slack API; tests pass a mocked ``WebClient`` to verify behaviour
    without touching the network.

    Args:
        message: The :class:`Message` to deliver.
    """
    client = self._get_client()
    channel = self._resolve_channel(message)
    # chat_postMessage is synchronous in slack_sdk; run it in a thread so
    # this stays friendly to the asyncio event loop.
    loop = asyncio.get_running_loop()
    await loop.run_in_executor(
        None,
        lambda: client.chat_postMessage(channel=channel, text=message.text()),
    )
    logger.debug("SlackAdapter.send -> channel=%s len=%d", channel, len(message.text()))

WebhookAdapter

Bases: MessagingAdapter

Messaging adapter that ingests messages via an HTTP webhook endpoint.

Incoming webhook payloads are validated, converted to :class:Message instances, and enqueued on an internal :class:asyncio.Queue. The :meth:receive async generator drains the queue and yields messages to consumers.

Typical usage alongside an HTTP framework::

adapter = WebhookAdapter()

@app.post("/webhook")
async def webhook(request):
    payload = await request.json()
    return await adapter.handle_webhook(payload)

async for message in adapter.receive():
    ...
Source code in src/ciel/gateway/adapter.py 
 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
128
129
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
class WebhookAdapter(MessagingAdapter):
    """Messaging adapter that ingests messages via an HTTP webhook endpoint.

    Incoming webhook payloads are validated, converted to :class:`Message`
    instances, and enqueued on an internal :class:`asyncio.Queue`. The
    :meth:`receive` async generator drains the queue and yields messages
    to consumers.

    Typical usage alongside an HTTP framework::

        adapter = WebhookAdapter()

        @app.post("/webhook")
        async def webhook(request):
            payload = await request.json()
            return await adapter.handle_webhook(payload)

        async for message in adapter.receive():
            ...
    """

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

    # -- public API -------------------------------------------------

    async def handle_webhook(self, payload: dict) -> dict:
        """Validate a webhook payload, enqueue a :class:`Message`, and respond.

        Expected payload shape::

            {"id": "...", "channel": "...", "sender": "...",
             "content": "...", "metadata": {...}}

        ``channel`` and ``content`` are required. ``id`` is generated
        via :func:`uuid.uuid4` when absent. ``sender`` defaults to
        ``None`` and ``metadata`` defaults to an empty dict.

        Args:
            payload: Raw deserialized webhook body.

        Returns:
            ``{"status": "accepted", "id": <message id>}`` on success,
            or ``{"status": "rejected", "error": "missing required
            field: <field>"}`` when validation fails.
        """
        channel = payload.get("channel")
        content = payload.get("content")

        if not channel:
            return {"status": "rejected", "error": "missing required field: channel"}
        if not content:
            return {"status": "rejected", "error": "missing required field: content"}

        message = Message(
            id=payload.get("id") or str(uuid.uuid4()),
            channel=channel,
            sender=payload.get("sender"),
            content=content,
            metadata=payload.get("metadata") or {},
        )

        await self._queue.put(message)

        return {"status": "accepted", "id": message.id}

    async def receive(self) -> AsyncIterator[Message]:
        """Async generator yielding messages from the internal queue.

        Blocks indefinitely on each :meth:`asyncio.Queue.get` call until
        a message becomes available, then yields it. The loop runs
        forever — the caller is responsible for cancellation/timeout.
        """
        while True:
            message = await self._queue.get()
            yield message

    async def receive_internal(self) -> Message:
        """Drain and return a single message from the queue.

        Convenience method for consumers that prefer one-at-a-time
        polling over the async generator. Blocks until a message is
        available.

        Returns:
            The next :class:`Message` in the queue.
        """
        return await self._queue.get()

    async def send(self, message: Message) -> None:
        """Outbound send — not implemented for the webhook adapter.

        The webhook adapter is inbound-only by design. Subclass or
        compose with an outbound adapter if delivery is needed.
        """
        raise NotImplementedError

handle_webhook(payload: dict) -> dict async

Validate a webhook payload, enqueue a :class:Message, and respond.

Expected payload shape::

{"id": "...", "channel": "...", "sender": "...",
 "content": "...", "metadata": {...}}

channel and content are required. id is generated via :func:uuid.uuid4 when absent. sender defaults to None and metadata defaults to an empty dict.

Parameters:

Name Type Description Default
payload dict

Raw deserialized webhook body.

 required

Returns:

Name Type Description
dict

{"status": "accepted", "id": <message id>} on success,
dict

or ``{"status": "rejected", "error": "missing required
field dict

"}`` when validation fails.
Source code in src/ciel/gateway/adapter.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
async def handle_webhook(self, payload: dict) -> dict:
    """Validate a webhook payload, enqueue a :class:`Message`, and respond.

    Expected payload shape::

        {"id": "...", "channel": "...", "sender": "...",
         "content": "...", "metadata": {...}}

    ``channel`` and ``content`` are required. ``id`` is generated
    via :func:`uuid.uuid4` when absent. ``sender`` defaults to
    ``None`` and ``metadata`` defaults to an empty dict.

    Args:
        payload: Raw deserialized webhook body.

    Returns:
        ``{"status": "accepted", "id": <message id>}`` on success,
        or ``{"status": "rejected", "error": "missing required
        field: <field>"}`` when validation fails.
    """
    channel = payload.get("channel")
    content = payload.get("content")

    if not channel:
        return {"status": "rejected", "error": "missing required field: channel"}
    if not content:
        return {"status": "rejected", "error": "missing required field: content"}

    message = Message(
        id=payload.get("id") or str(uuid.uuid4()),
        channel=channel,
        sender=payload.get("sender"),
        content=content,
        metadata=payload.get("metadata") or {},
    )

    await self._queue.put(message)

    return {"status": "accepted", "id": message.id}

receive() -> AsyncIterator[Message] async

Async generator yielding messages from the internal queue.

Blocks indefinitely on each :meth:asyncio.Queue.get call until a message becomes available, then yields it. The loop runs forever — the caller is responsible for cancellation/timeout.

Source code in src/ciel/gateway/adapter.py 
150
151
152
153
154
155
156
157
158
159
async def receive(self) -> AsyncIterator[Message]:
    """Async generator yielding messages from the internal queue.

    Blocks indefinitely on each :meth:`asyncio.Queue.get` call until
    a message becomes available, then yields it. The loop runs
    forever — the caller is responsible for cancellation/timeout.
    """
    while True:
        message = await self._queue.get()
        yield message

receive_internal() -> Message async

Drain and return a single message from the queue.

Convenience method for consumers that prefer one-at-a-time polling over the async generator. Blocks until a message is available.

Returns:

Type Description
Message

The next :class:Message in the queue.
Source code in src/ciel/gateway/adapter.py 
161
162
163
164
165
166
167
168
169
170
171
async def receive_internal(self) -> Message:
    """Drain and return a single message from the queue.

    Convenience method for consumers that prefer one-at-a-time
    polling over the async generator. Blocks until a message is
    available.

    Returns:
        The next :class:`Message` in the queue.
    """
    return await self._queue.get()

send(message: Message) -> None async

Outbound send — not implemented for the webhook adapter.

The webhook adapter is inbound-only by design. Subclass or compose with an outbound adapter if delivery is needed.

Source code in src/ciel/gateway/adapter.py 
173
174
175
176
177
178
179
async def send(self, message: Message) -> None:
    """Outbound send — not implemented for the webhook adapter.

    The webhook adapter is inbound-only by design. Subclass or
    compose with an outbound adapter if delivery is needed.
    """
    raise NotImplementedError

create_control_app(*, runtime: DefaultAgentRuntime, registry: Optional[ProviderRegistry] = None, audit_sink: Any = None, tenant_id: Optional[str] = None, api_key: Optional[str] = None, board_db_path: Optional[str] = None) -> FastAPI

Build a FastAPI control plane for the Ciel agent runtime.

Parameters

runtime: A :class:ciel.runtime.DefaultAgentRuntime used to run agent loops and dispatch tools. Tool dispatch goes through runtime.dispatcher. El endpoint POST /v1/agent/run/stream usa runtime.stream_tokens para emitir Server-Sent Events (SSE) con los fragmentos incrementales del assistant y cierra con data: [DONE]. registry: Optional :class:ciel.providers.ProviderRegistry surfaced via the GET /info endpoint. If omitted, an empty provider list is reported. audit_sink: Optional audit sink retained for observability hooks. Not directly invoked here — the runtime already emits audit events. tenant_id: Default tenant propagated to the runtime when a request does not provide its own tenant_id. api_key: Optional transport API key. When None (the default), the CIEL_API_KEY environment variable is consulted. If a key is configured (explicitly or via env), protected routes require a valid key via Authorization: Bearer *** orX-API-Key; otherwise the dependency is a no-op (open mode). board_db_path: Optional path to a SQLite database for the kanban board. When set (o bien vía la variable de entornoCIEL_BOARD_DB), el board se monta sobre SQLite para que/v1/board/list`` vea las tareas creadas por el CLI y viceversa. Si no se configura ninguna ruta, el board queda en memoria (comportamiento legacy, útil para smoke tests/offline).

Source code in src/ciel/gateway/base.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
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
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
def create_control_app(
    *,
    runtime: DefaultAgentRuntime,
    registry: Optional[ProviderRegistry] = None,
    audit_sink: Any = None,
    tenant_id: Optional[str] = None,
    api_key: Optional[str] = None,
    board_db_path: Optional[str] = None,
) -> FastAPI:
    """Build a FastAPI control plane for the Ciel agent runtime.

    Parameters
    ----------
    runtime:
        A :class:`ciel.runtime.DefaultAgentRuntime` used to run agent loops and
        dispatch tools. Tool dispatch goes through ``runtime.dispatcher``.
        El endpoint ``POST /v1/agent/run/stream`` usa
        ``runtime.stream_tokens`` para emitir Server-Sent Events (SSE) con los
        fragmentos incrementales del assistant y cierra con ``data: [DONE]``.
    registry:
        Optional :class:`ciel.providers.ProviderRegistry` surfaced via the
        ``GET /info`` endpoint. If omitted, an empty provider list is reported.
    audit_sink:
        Optional audit sink retained for observability hooks. Not directly
        invoked here — the runtime already emits audit events.
    tenant_id:
        Default tenant propagated to the runtime when a request does not
        provide its own ``tenant_id``.
    api_key:
        Optional transport API key. When ``None`` (the default), the
        ``CIEL_API_KEY`` environment variable is consulted. If a key is
        configured (explicitly or via env), protected routes require a valid
        key via ``Authorization: Bearer *** or ``X-API-Key``; otherwise
        the dependency is a no-op (open mode).
    board_db_path:
        Optional path to a SQLite database for the kanban board. When set (o
        bien vía la variable de entorno ``CIEL_BOARD_DB``), el board se monta
        sobre SQLite para que ``/v1/board/list`` vea las tareas creadas por el
        CLI y viceversa. Si no se configura ninguna ruta, el board queda en
        memoria (comportamiento legacy, útil para smoke tests/offline).
    """
    from ciel.gateway.auth import Depends, make_auth_dependency

    api_key_guard = Depends(make_auth_dependency(expected_key=api_key))

    app = FastAPI(
        title="Ciel Control Gateway",
        version=__version__,
        description="HTTP control plane for the Ciel agent runtime.",
    )
    # Attach references for tests / introspection.
    app.state.runtime = runtime
    app.state.registry = registry
    app.state.audit_sink = audit_sink
    app.state.default_tenant_id = tenant_id

    # Resolve a live board instance if orchestration is available.
    # Si se pasa board_db_path o existe CIEL_BOARD_DB, el board se monta sobre
    # SQLite para compartir estado con el CLI; en caso contrario queda en
    # memoria (fallback para smoke tests/offline).
    board: Optional[KanbanBoard] = None
    resolved_board_db = board_db_path or os.environ.get("CIEL_BOARD_DB")
    try:
        board = KanbanBoard(path=resolved_board_db)  # None => en memoria.
    except Exception:  # pragma: no cover - defensive guard
        board = None
    app.state.board = board

    def _resolve_tenant(request_tenant: Optional[str]) -> Optional[str]:
        return request_tenant or tenant_id

    @app.get("/health", response_model=HealthResponse, tags=["meta"])
    async def health() -> HealthResponse:
        return HealthResponse(version=__version__)

    # --- Fase 14 / F16: health reales (liveness vs readiness) ---------------
    @app.get("/healthz", response_model=LivenessResponse, tags=["meta"])
    async def healthz() -> LivenessResponse:
        # Liveness: el proceso está vivo. No depende de backends remotos.
        return LivenessResponse(status="alive", version=__version__)

    @app.get("/readyz", response_model=ReadinessResponse, tags=["meta"])
    async def readyz(request: Request) -> ReadinessResponse:
        # Readiness: el gateway puede atender tráfico. Depende de que el
        # StateBackend compartido esté conectado y migrado.
        backend = getattr(request.app.state, "state_backend", None)
        if backend is None:
            # Sin backend cableado => no listo para multi-réplica.
            return ReadinessResponse(
                status="not_ready",
                version=__version__,
                backend="none",
                backend_ready=False,
                details={"reason": "state_backend not wired"},
            )
        ready = backend.is_ready()
        return ReadinessResponse(
            status="ready" if ready else "not_ready",
            version=__version__,
            backend=backend.backend_type,
            backend_ready=ready,
            details={"ready": ready},
        )

    @app.get("/info", response_model=InfoResponse, tags=["meta"])
    async def info() -> InfoResponse:
        providers: List[ProviderInfo] = []
        if registry is not None:
            for name in registry.available():
                provider_name: Optional[str] = None
                try:
                    provider = registry.get(name)
                    provider_name = getattr(provider, "provider_name", None)
                except Exception:
                    provider_name = None
                providers.append(ProviderInfo(name=name, provider_name=provider_name))
        return InfoResponse(
            version=__version__,
            providers=providers,
            default_tenant=tenant_id,
        )

    @app.post(
        "/v1/agent/run",
        response_model=AgentRunResponse,
        tags=["agent"],
        dependencies=[api_key_guard],
    )
    async def agent_run(body: AgentRunRequest) -> AgentRunResponse:
        effective_tenant = _resolve_tenant(body.tenant_id)
        if effective_tenant is None:
            raise HTTPException(
                status_code=400,
                detail="tenant_id is required (multi-tenancy is enforced). "
                "Provide it in the request or configure a default tenant in the gateway.",
            )
        session_id = body.session_id or str(uuid.uuid4())
        request = _build_chat_request(body, session_id, effective_tenant)
        try:
            result = await runtime.run_agent_loop(
                request=request,
                tenant_id=effective_tenant,
                toolset=body.toolset,
            )
        except Exception as exc:
            logger.exception("agent.run failed: %s", exc)
            raise HTTPException(status_code=500, detail=f"agent loop failed: {exc}") from exc
        return _serialize_chat_response(result)

    def _build_chat_request(body: AgentRunRequest, session_id: str, effective_tenant: Optional[str]) -> ChatRequest:
        extra: Dict[str, Any] = dict(body.extra or {})
        extra.setdefault("session_id", session_id)
        if effective_tenant is not None:
            extra.setdefault("tenant_id", effective_tenant)
        return ChatRequest(
            messages=(ChatMessage(role="user", content=body.prompt),),
            tools=(),
            model=body.model,
            temperature=body.temperature,
            max_tokens=body.max_tokens,
            extra=extra,
        )

    @app.post(
        "/v1/agent/run/stream",
        tags=["agent"],
        dependencies=[api_key_guard],
    )
    async def agent_run_stream(body: AgentRunRequest) -> StreamingResponse:
        """Stream the assistant completion as Server-Sent Events.

        Emite cada fragmento incremental del assistant como un evento SSE
        ``data: <token>`` y cierra la respuesta con ``data: [DONE]``.
        Reutiliza ``runtime.stream_tokens`` (que a su vez llama a
        ``provider.stream`` para hacer SSE real cuando el proveedor lo soporta).

        El ``tenant_id`` es obligatorio, igual que en ``/v1/agent/run``.
        """
        effective_tenant = _resolve_tenant(body.tenant_id)
        if effective_tenant is None:
            raise HTTPException(
                status_code=400,
                detail="tenant_id is required (multi-tenancy is enforced). "
                "Provide it in the request or configure a default tenant in the gateway.",
            )
        session_id = body.session_id or str(uuid.uuid4())

        async def _event_generator():
            try:
                request = _build_chat_request(body, session_id, effective_tenant)
                async for token in runtime.stream_tokens(
                    request=request,
                    tenant_id=effective_tenant,
                    toolset=body.toolset,
                ):
                    yield f"data: {token}\n\n"
            except Exception as exc:  # pragma: no cover - defensive guard
                logger.exception("agent.run.stream failed: %s", exc)
                yield f"data: [ERROR] {exc}\n\n"
            yield "data: [DONE]\n\n"

        return StreamingResponse(_event_generator(), media_type="text/event-stream")

    @app.post(
        "/v1/tools/{toolset}/{name}",
        response_model=ToolInvokeResponse,
        tags=["tools"],
        dependencies=[api_key_guard],
    )
    async def tool_invoke(toolset: str, name: str, body: ToolInvokeRequest) -> ToolInvokeResponse:
        dispatcher: Optional[DefaultToolDispatcher] = getattr(runtime, "dispatcher", None)
        if dispatcher is None:
            raise HTTPException(status_code=500, detail="runtime has no tool dispatcher")
        effective_tenant = _resolve_tenant(body.tenant_id)
        if effective_tenant is None:
            raise HTTPException(
                status_code=400,
                detail="tenant_id is required to dispatch tools (multi-tenancy is enforced).",
            )
        tool_call_id = body.tool_call_id or str(uuid.uuid4())
        try:
            result = await dispatcher.dispatch(
                tenant_id=effective_tenant,
                toolset=toolset,
                name=name,
                arguments=dict(body.arguments or {}),
                tool_call_id=tool_call_id,
            )
        except Exception as exc:
            logger.exception("tool.invoke failed: %s", exc)
            raise HTTPException(status_code=500, detail=f"tool dispatch failed: {exc}") from exc
        return ToolInvokeResponse(
            id=getattr(result, "id", None),
            name=getattr(result, "name", name),
            output=getattr(result, "output", None),
            error=getattr(result, "error", None),
            metadata=dict(getattr(result, "metadata", {}) or {}),
        )

    @app.get(
        "/v1/board/list",
        response_model=BoardListResponse,
        tags=["board"],
        dependencies=[api_key_guard],
    )
    async def board_list(tenant_id: Optional[str] = None) -> BoardListResponse:
        effective_tenant = _resolve_tenant(tenant_id)
        if board is None:
            return BoardListResponse(tasks=[])
        return BoardListResponse(tasks=_list_board_tasks(board, effective_tenant))

    return app

create_discord_webhook_router(adapter, *, path: str = '/v1/messaging/discord') -> APIRouter

Router que ingiere eventos MESSAGE_CREATE de Discord al adapter.

Discord entrega un JSON con content, author/id y channel_id. El router lo convierte en Message y lo enqueuea.

Source code in src/ciel/gateway/messaging.py 
 72
 73
 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
def create_discord_webhook_router(
    adapter,
    *,
    path: str = "/v1/messaging/discord",
) -> APIRouter:
    """Router que ingiere eventos ``MESSAGE_CREATE`` de Discord al ``adapter``.

    Discord entrega un JSON con ``content``, ``author``/``id`` y ``channel_id``.
    El router lo convierte en ``Message`` y lo enqueuea.
    """
    router = APIRouter()

    @router.post(path)
    async def discord_events(request: Request):
        try:
            payload = await request.json()
        except Exception as exc:
            return {"status": "rejected", "error": f"invalid json: {exc}"}

        if payload.get("type") == 1:
            # Discord Ping (interacción): responde con el mismo id.
            return {"type": 1}

        content = payload.get("content", "")
        if not content:
            return {"status": "ok"}  # eventos sin contenido (presence, etc.)
        author = payload.get("author", {})
        sender = author.get("username") or author.get("id")
        message = Message(
            id=payload.get("id") or str(__import__("uuid").uuid4()),
            channel="discord",
            sender=sender,
            content=content,
            metadata={"discord_event": payload, "channel_id": payload.get("channel_id")},
        )
        enqueue = getattr(adapter, "enqueue", None)
        if callable(enqueue):
            enqueue(message)
        else:  # pragma: no cover - adapter sin enqueue
            logger.warning("DiscordAdapter has no enqueue(); dropping %s", message.id)
        return {"status": "accepted", "id": message.id}

    @router.get(f"{path}/health")
    async def discord_health():
        return {"status": "ok", "channel": "discord"}

    return router

create_slack_webhook_router(adapter: 'SlackAdapter', *, path: str = '/v1/messaging/slack', signing_secret: str | None = None)

Build an APIRouter that ingests inbound Slack events into adapter.

Slack delivers events (url_verification challenges and event_callback payloads) as JSON POST requests. This router:

  • answers the url_verification challenge so the Slack app can be installed/verified, and
  • for message events, converts the Slack event into a :class:~ciel.gateway.adapter.Message and enqueues it for the agent.

The adapter is expected to expose an enqueue method (a subclass of :class:SlackAdapter that also holds an internal queue — e.g. via composition with :class:~ciel.gateway.adapter.WebhookAdapter) so events can be drained by the runtime. To keep coupling low, this router calls adapter.enqueue(...) when present and otherwise logs.

Parameters

adapter: A :class:SlackAdapter (or compatible) that should receive inbound events. path: Route under which the Slack endpoint is exposed. signing_secret: Optional Slack signing secret used for request verification. When provided, payloads are verified; when omitted, verification is skipped (acceptable for local/offline smoke testing).

Source code in src/ciel/gateway/__init__.py 
 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
128
129
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
def create_slack_webhook_router(
    adapter: "SlackAdapter",  # noqa: F821
    *,
    path: str = "/v1/messaging/slack",
    signing_secret: str | None = None,
):
    """Build an APIRouter that ingests inbound Slack events into ``adapter``.

    Slack delivers events (``url_verification`` challenges and
    ``event_callback`` payloads) as JSON ``POST`` requests. This router:

    * answers the ``url_verification`` challenge so the Slack app can be
      installed/verified, and
    * for ``message`` events, converts the Slack event into a
      :class:`~ciel.gateway.adapter.Message` and enqueues it for the agent.

    The adapter is expected to expose an ``enqueue`` method (a subclass of
    :class:`SlackAdapter` that also holds an internal queue — e.g. via
    composition with :class:`~ciel.gateway.adapter.WebhookAdapter`) so events
    can be drained by the runtime. To keep coupling low, this router calls
    ``adapter.enqueue(...)`` when present and otherwise logs.

    Parameters
    ----------
    adapter:
        A :class:`SlackAdapter` (or compatible) that should receive inbound
        events.
    path:
        Route under which the Slack endpoint is exposed.
    signing_secret:
        Optional Slack signing secret used for request verification. When
        provided, payloads are verified; when omitted, verification is skipped
        (acceptable for local/offline smoke testing).
    """
    from fastapi import APIRouter, Request

    router = APIRouter()

    @router.post(path)
    async def slack_events(request: Request):
        try:
            payload = await request.json()
        except Exception as exc:
            return {"status": "rejected", "error": f"invalid json: {exc}"}

        # Slack URL verification handshake.
        if payload.get("type") == "url_verification":
            return {"challenge": payload.get("challenge")}

        event = payload.get("event", {})
        if event.get("type") == "message" and not event.get("subtype"):
            channel = event.get("channel")
            message = Message(
                id=event.get("client_msg_id") or event.get("ts") or "",
                channel=channel,
                sender=event.get("user"),
                content=event.get("text", ""),
                metadata={"slack_event": event, "team": payload.get("team_id")},
            )
            enqueue = getattr(adapter, "enqueue", None)
            if callable(enqueue):
                # SlackAdapter.enqueue is synchronous (put_nowait); do NOT await it.
                enqueue(message)
            else:  # pragma: no cover - adapter without enqueue
                logger.warning(
                    "SlackAdapter has no enqueue(); dropping inbound message %s", message.id
                )
            return {"status": "accepted", "id": message.id}

        # Other event types (app_mention, etc.) — acknowledge to avoid retries.
        return {"status": "ok"}

    @router.get(f"{path}/health")
    async def slack_health():
        return {"status": "ok", "channel": "slack"}

    return router

create_teams_webhook_router(adapter, *, path: str = '/v1/messaging/teams', signing_secret: Optional[str] = None) -> APIRouter

Router que ingiere eventos de Microsoft Teams al adapter.

Teams entrega cargas JSON con text (y opcionalmente from / channelData). El router los convierte en Message y los enqueuea. Si se pasa signing_secret se puede validar la firma HMAC (fuera de alcance en smoke tests); sin él, se acepta el payload tal cual.

Source code in src/ciel/gateway/messaging.py 
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
64
65
66
67
68
69
def create_teams_webhook_router(
    adapter,
    *,
    path: str = "/v1/messaging/teams",
    signing_secret: Optional[str] = None,
) -> APIRouter:
    """Router que ingiere eventos de Microsoft Teams al ``adapter``.

    Teams entrega cargas JSON con ``text`` (y opcionalmente ``from`` /
    ``channelData``). El router los convierte en ``Message`` y los enqueuea.
    Si se pasa ``signing_secret`` se puede validar la firma HMAC (fuera de
    alcance en smoke tests); sin él, se acepta el payload tal cual.
    """
    router = APIRouter()

    @router.post(path)
    async def teams_events(request: Request):
        try:
            payload = await request.json()
        except Exception as exc:
            return {"status": "rejected", "error": f"invalid json: {exc}"}

        text = payload.get("text") or payload.get("body") or ""
        if not text:
            return {"status": "rejected", "error": "missing text"}
        sender = payload.get("from", {}).get("id") if isinstance(payload.get("from"), dict) else payload.get("from")
        message = Message(
            id=payload.get("id") or str(__import__("uuid").uuid4()),
            channel="teams",
            sender=sender,
            content=text,
            metadata={"teams_event": payload},
        )
        enqueue = getattr(adapter, "enqueue", None)
        if callable(enqueue):
            enqueue(message)
        else:  # pragma: no cover - adapter sin enqueue
            logger.warning("TeamsAdapter has no enqueue(); dropping %s", message.id)
        return {"status": "accepted", "id": message.id}

    @router.get(f"{path}/health")
    async def teams_health():
        return {"status": "ok", "channel": "teams"}

    return router

create_webhook_router(adapter: WebhookAdapter, *, path: str = '/v1/messaging/webhook', api_key: Optional[str] = None)

Build an APIRouter that pushes inbound HTTP messages into adapter.

The router exposes POST {path} (accept a message) and GET {path}/health. It is intentionally decoupled from a concrete agent runtime so it can be composed by any surface (control gateway, ACP, MCP).

Source code in src/ciel/gateway/__init__.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def create_webhook_router(
    adapter: WebhookAdapter,
    *,
    path: str = "/v1/messaging/webhook",
    api_key: Optional[str] = None,
):
    """Build an APIRouter that pushes inbound HTTP messages into ``adapter``.

    The router exposes ``POST {path}`` (accept a message) and
    ``GET {path}/health``. It is intentionally decoupled from a concrete agent
    runtime so it can be composed by any surface (control gateway, ACP, MCP).
    """
    from fastapi import APIRouter

    from ciel.gateway.auth import Depends, make_auth_dependency

    api_key_guard = Depends(make_auth_dependency(expected_key=api_key))

    router = APIRouter()

    @router.post(path, dependencies=[api_key_guard])
    async def receive_webhook(request: Request):
        try:
            payload = await request.json()
        except Exception as exc:
            return {"status": "rejected", "error": f"invalid json: {exc}"}
        return await adapter.handle_webhook(payload)

    @router.get(f"{path}/health")
    async def webhook_health():
        return {"status": "ok", "channel": "webhook"}

    return router

create_webui_router(adapter, *, path: str = '/v1/messaging/webui') -> APIRouter

Router de Web UI: POST para enviar mensajes al runtime, GET para sondear.

La UI hace POST {path} con un Message y lo enqueuea en el WebUIAdapter; GET {path}/outbound sondea el siguiente mensaje saliente del adapter (polling).

Source code in src/ciel/gateway/messaging.py 
121
122
123
124
125
126
127
128
129
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
def create_webui_router(
    adapter,
    *,
    path: str = "/v1/messaging/webui",
) -> APIRouter:
    """Router de Web UI: POST para enviar mensajes al runtime, GET para sondear.

    La UI hace ``POST {path}`` con un ``Message`` y lo enqueuea en el
    ``WebUIAdapter``; ``GET {path}/outbound`` sondea el siguiente mensaje
    saliente del adapter (polling).
    """
    from ciel.adapters import WebUIAdapter

    router = APIRouter()

    @router.post(path)
    async def webui_inbound(request: Request):
        try:
            payload = await request.json()
        except Exception as exc:
            return {"status": "rejected", "error": f"invalid json: {exc}"}
        content = payload.get("content")
        if not content:
            return {"status": "rejected", "error": "missing content"}
        message = Message(
            id=payload.get("id") or str(__import__("uuid").uuid4()),
            channel="webui",
            sender=payload.get("sender"),
            content=content,
            metadata=payload.get("metadata") or {},
        )
        enqueue = getattr(adapter, "enqueue", None)
        if callable(enqueue):
            enqueue(message)
        else:  # pragma: no cover - adapter sin enqueue
            logger.warning("WebUIAdapter has no enqueue(); dropping %s", message.id)
        return {"status": "accepted", "id": message.id}

    @router.get(f"{path}/outbound")
    async def webui_outbound():
        # Polling: devuelve el siguiente mensaje saliente o 204 si no hay.
        if not isinstance(adapter, WebUIAdapter):
            return {"status": "ok", "channel": "webui"}
        try:
            message = adapter._outbound.get_nowait()
        except Exception:
            return {"status": "no_content"}
        return {
            "status": "ok",
            "id": message.id,
            "content": message.text(),
            "sender": message.sender,
            "metadata": message.metadata,
        }

    @router.get(f"{path}/health")
    async def webui_health():
        return {"status": "ok", "channel": "webui"}

    return router

make_app(*, tenant_id: Optional[str] = None, include_mcp: bool = True, include_webhook: bool = True, state_backend: Optional[object] = None) -> FastAPI

Compose the full Ciel gateway into a single FastAPI app.

Parameters

tenant_id: Default tenant propagated to the runtime when a request omits its own tenant_id. Falls back to the CIEL_TENANT environment variable. include_mcp: Mount the MCP host app at /mcp. include_webhook: Mount the webhook messaging router at /v1/messaging/webhook. state_backend: Optional pre-built :class:StateBackend (Fase 14 / F15). When omitted, one is built from CIEL_STATE_BACKEND / CIEL_STATE_DSN (default SQLite). Injecting a shared backend lets tests and multi-replica deployments reuse the same state source.

Source code in src/ciel/gateway/server.py 
 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
128
129
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
181
182
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
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
def make_app(
    *,
    tenant_id: Optional[str] = None,
    include_mcp: bool = True,
    include_webhook: bool = True,
    state_backend: Optional[object] = None,
) -> FastAPI:
    """Compose the full Ciel gateway into a single FastAPI app.

    Parameters
    ----------
    tenant_id:
        Default tenant propagated to the runtime when a request omits its own
        ``tenant_id``. Falls back to the ``CIEL_TENANT`` environment variable.
    include_mcp:
        Mount the MCP host app at ``/mcp``.
    include_webhook:
        Mount the webhook messaging router at ``/v1/messaging/webhook``.
    state_backend:
        Optional pre-built :class:`StateBackend` (Fase 14 / F15). When omitted,
        one is built from ``CIEL_STATE_BACKEND`` / ``CIEL_STATE_DSN`` (default
        SQLite). Injecting a shared backend lets tests and multi-replica
        deployments reuse the same state source.
    """
    tenant_id = tenant_id or os.getenv("CIEL_TENANT")

    # --- runtime wiring ----------------------------------------------------
    chat_provider = _build_chat_provider()
    registry = ToolRegistry(default_toolset="default")
    tool_provider = ToolProvider(registry=registry, require_tenant_on_execution=True)
    dispatcher = DefaultToolDispatcher(provider=tool_provider, default_toolset="default")
    runtime = DefaultAgentRuntime(provider=chat_provider, dispatcher=dispatcher)

    # --- Fase 14 / F15: StateBackend compartido (multi-réplica) ------------
    # Construye el backend de estado según CIEL_STATE_BACKEND / CIEL_STATE_DSN
    # (default: SQLite local offline-safe). Los stores de checkpoint/session
    # (F16) lo consumen para compartir estado entre réplicas. Expuesto en
    # app.state para los health endpoints (/readyz) y resume-on-startup.
    from ciel.runtime.state_backend import StateBackend, build_state_backend

    if state_backend is None:
        state_backend = build_state_backend()
    else:
        # Validación defensiva: debe cumplir la interfaz StateBackend.
        assert isinstance(state_backend, StateBackend), "state_backend debe ser un StateBackend"


    # Lazy import to avoid a circular import with the ``ciel.gateway`` package
    # (server.py is imported by gateway/__init__.py; the two helpers below are
    # defined there and only needed at call time, after the package is loaded).
    from ciel.gateway import create_webhook_router, mount_mcp_app

    # --- control plane (root app) -----------------------------------------
    app = create_control_app(runtime=runtime, registry=None, tenant_id=tenant_id)
    app.title = "Ciel Agent Framework Gateway"
    app.version = __version__

    # --- MCP host (mounted sub-application) -------------------------------
    if include_mcp:
        # Expose the JSON-RPC endpoint at the mount root so the resulting route
        # is ``POST /mcp/`` and health at ``GET /mcp/health``.
        mcp_app = mount_mcp_app(dispatcher=dispatcher, tenant_id=tenant_id, path="/")
        app.mount("/mcp", mcp_app)
        app.state.mcp_mounted = True
    else:
        app.state.mcp_mounted = False

    # --- webhook messaging adapter ----------------------------------------
    if include_webhook:
        adapter = WebhookAdapter()
        app.include_router(create_webhook_router(adapter))
        app.state.webhook_adapter = adapter

    # --- Fase 8: adapters de canal Teams/Discord/Web UI (offline-safe) -----
    # Se montan solo si no estamos en modo ultra-ligero (siempre por defecto).
    # Cada router enqueuea eventos entrantes en su adapter; el runtime los
    # consume vía ``adapter.receive()``. No requieren red para arrancar.
    try:
        from ciel.adapters import DiscordAdapter, TeamsAdapter, WebUIAdapter
        from ciel.gateway.messaging import (
            create_discord_webhook_router,
            create_teams_webhook_router,
            create_webui_router,
        )

        teams_adapter = TeamsAdapter(webhook_url=os.getenv("CIEL_TEAMS_WEBHOOK"))
        discord_adapter = DiscordAdapter(webhook_url=os.getenv("CIEL_DISCORD_WEBHOOK"))
        webui_adapter = WebUIAdapter()
        app.include_router(create_teams_webhook_router(teams_adapter))
        app.include_router(create_discord_webhook_router(discord_adapter))
        app.include_router(create_webui_router(webui_adapter))
        app.state.teams_adapter = teams_adapter
        app.state.discord_adapter = discord_adapter
        app.state.webui_adapter = webui_adapter
        logger.info("ciel serve: mounted Teams/Discord/WebUI messaging routers")
    except Exception as exc:  # pragma: no cover - defensive
        logger.warning("ciel serve: channel adapters not mounted: %s", exc)

    # --- Fase 13 / F19: Ciel Studio dashboard (offline-safe) --------
    # Expone /v1/studio con el snapshot de sesiones/loops. Usa el store
    # singleton de studio (compartido con install_studio_support).
    try:
        from ciel.studio import create_studio_router, get_studio_store

        studio_store = get_studio_store()
        app.include_router(create_studio_router(store=studio_store))
        app.state.studio_store = studio_store
        logger.info("ciel serve: mounted Ciel Studio dashboard at /v1/studio")
    except Exception as exc:  # pragma: no cover - defensive
        logger.warning("ciel serve: Ciel Studio not mounted: %s", exc)

    # --- Fase 13 / F20: Ciel Studio graph trace + replay ------------
    try:
        from ciel.studio_trace import create_trace_router, get_trace_store

        trace_store = get_trace_store()
        app.include_router(create_trace_router(store=trace_store))
        app.state.trace_store = trace_store
        logger.info("ciel serve: mounted Ciel Studio trace at /v1/studio/trace")
    except Exception as exc:  # pragma: no cover - defensive
        logger.warning("ciel serve: Ciel Studio trace not mounted: %s", exc)

    # --- Fase 13 / F21: Ciel Studio cost dashboard -------------------
    try:
        from ciel.studio_cost import create_cost_router, get_cost_store

        cost_store = get_cost_store()
        app.include_router(create_cost_router(store=cost_store))
        app.state.cost_store = cost_store
        logger.info("ciel serve: mounted Ciel Studio cost at /v1/studio/cost")
    except Exception as exc:  # pragma: no cover - defensive
        logger.warning("ciel serve: Ciel Studio cost not mounted: %s", exc)

    # --- Prometheus /metrics endpoint (lenient) --------------------------
    # Only mounted when prometheus-client is available; otherwise the import
    # is skipped so the gateway still boots offline.
    try:
        from ciel.observability.metrics import PROM_AVAILABLE, metrics_handler

        if PROM_AVAILABLE:
            app.add_api_route(
                "/metrics", metrics_handler, methods=["GET"], include_in_schema=False
            )
            app.state.metrics_mounted = True
            logger.info("ciel serve: mounted /metrics (Prometheus)")
        else:  # pragma: no cover - depends on optional extra
            app.state.metrics_mounted = False
            logger.info("ciel serve: /metrics not mounted (prometheus-client absent)")
    except Exception as exc:  # pragma: no cover - defensive
        app.state.metrics_mounted = False
        logger.warning("ciel serve: /metrics mount skipped: %s", exc)

    # --- Fase 14 / F15: stores de resume sobre el StateBackend compartido -
    # Instancia los stores de checkpoint/session contra el backend para que
    # estén disponibles para resume-on-startup (F16) y otros componentes, sin
    # alterar el control plane existente. Multi-réplica: comparten state vía
    # el backend (Postgres en prod, SQLite local en dev).
    try:
        from ciel.orchestration.agent import EventLoopCheckpointStore
        from ciel.orchestration.graph import GraphCheckpointStore
        from ciel.orchestration.session import SessionStore
        from ciel.runtime.checkpoints import CheckpointStore

        checkpoint_store = CheckpointStore(state_backend)
        session_store = SessionStore(state_backend)
        graph_checkpoint_store = GraphCheckpointStore(state_backend)
        eventloop_checkpoint_store = EventLoopCheckpointStore(state_backend)
        app.state.state_backend = state_backend
        app.state.checkpoint_store = checkpoint_store
        app.state.session_store = session_store
        app.state.graph_checkpoint_store = graph_checkpoint_store
        app.state.eventloop_checkpoint_store = eventloop_checkpoint_store
        logger.info(
            "ciel serve: state backend '%s' listo (stores de resume instanciados)",
            state_backend.backend_type,
        )
    except Exception as exc:  # pragma: no cover - defensive
        app.state.state_backend = state_backend
        logger.warning("ciel serve: state stores not wired: %s", exc)

    return app

mount_mcp_app(*, dispatcher=None, tenant_id: str | None = None, path: str = '/mcp') -> 'FastAPI'

Build a FastAPI app that serves the Ciel MCP host endpoint.

The endpoint at {path} accepts MCP JSON-RPC payloads and routes them to a :class:MCPServer wired to the runtime's tool provider. Callers normally mount this under the control gateway or run it as a standalone service.

Parameters

dispatcher: Optional :class:ciel.runtime.DefaultToolDispatcher. When omitted, a bare :class:MCPServer with no provider is created (logs initialize / shutdown only). tenant_id: Default tenant used for audit propagation. path: Route under which the MCP endpoint is exposed.

Source code in src/ciel/gateway/__init__.py 
176
177
178
179
180
181
182
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
210
211
212
213
214
215
216
217
218
219
220
221
def mount_mcp_app(
    *,
    dispatcher=None,
    tenant_id: str | None = None,
    path: str = "/mcp",
) -> "FastAPI":  # noqa: F821
    """Build a FastAPI app that serves the Ciel MCP host endpoint.

    The endpoint at ``{path}`` accepts MCP JSON-RPC payloads and routes them to
    a :class:`MCPServer` wired to the runtime's tool provider. Callers normally
    mount this under the control gateway or run it as a standalone service.

    Parameters
    ----------
    dispatcher:
        Optional :class:`ciel.runtime.DefaultToolDispatcher`. When omitted, a
        bare :class:`MCPServer` with no provider is created (logs initialize /
        shutdown only).
    tenant_id:
        Default tenant used for audit propagation.
    path:
        Route under which the MCP endpoint is exposed.
    """
    from fastapi import FastAPI

    from ciel.gateway.mcp import MCPServer

    provider = getattr(dispatcher, "provider", None) if dispatcher is not None else None
    server = MCPServer(provider=provider, tenant_id=tenant_id)

    app = FastAPI(title="Ciel MCP Host", version="0.1.0")
    app.state.mcp = server

    @app.post(path)
    async def mcp_endpoint(request: Request):
        try:
            payload = await request.json()
        except Exception as exc:
            return {"jsonrpc": "2.0", "id": None, "error": {"code": -32700, "message": f"parse error: {exc}"}}
        return await server.handle(payload)

    @app.get("/health")
    async def health():
        return {"status": "ok", "service": "ciel-mcp", "version": "0.1.0"}

    return app

__all__ = ['create_control_app'] module-attribute

__version__ = _pkg_version('ciel') module-attribute

logger = logging.getLogger(__name__) module-attribute

AgentRunRequest

Bases: BaseModel

Source code in src/ciel/gateway/base.py 
63
64
65
66
67
68
69
70
71
class AgentRunRequest(BaseModel):
    prompt: str = Field(..., description="User prompt text.")
    model: Optional[str] = Field(None, description="Model id to use for completion.")
    toolset: Optional[str] = Field(None, description="Toolset name for tool dispatch.")
    tenant_id: Optional[str] = Field(None, description="Tenant identifier propagated to runtime.")
    session_id: Optional[str] = Field(None, description="Optional session id for tracing.")
    temperature: Optional[float] = None
    max_tokens: Optional[int] = None
    extra: Dict[str, Any] = Field(default_factory=dict)

AgentRunResponse

Bases: BaseModel

Source code in src/ciel/gateway/base.py 
82
83
84
85
class AgentRunResponse(BaseModel):
    text: str
    session_id: Optional[str] = None
    tool_results: List[ToolResultPayload]

AgentRuntimeResult dataclass

Source code in src/ciel/runtime/tools.py 
216
217
218
219
220
221
@dataclass(frozen=True)
class AgentRuntimeResult:
    response: ChatResponse
    loop_results: Sequence[ToolLoopResult]
    tenant_id: Optional[str] = None
    metadata: Dict[str, Any] = field(default_factory=dict)

BoardListResponse

Bases: BaseModel

Source code in src/ciel/gateway/base.py 
111
112
class BoardListResponse(BaseModel):
    tasks: List[BoardTaskPayload]

BoardTaskPayload

Bases: BaseModel

Source code in src/ciel/gateway/base.py 
102
103
104
105
106
107
108
class BoardTaskPayload(BaseModel):
    id: str
    title: str
    status: str = "todo"
    assignee: Optional[str] = None
    tenant_id: Optional[str] = None
    metadata: Dict[str, Any] = Field(default_factory=dict)

ChatMessage dataclass

Source code in src/ciel/runtime/tools.py 
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
@dataclass(frozen=True)
class ChatMessage:
    role: str
    content: ChatContent
    name: Optional[str] = None
    tool_call_id: Optional[str] = None
    tool_calls: Optional[list[dict[str, Any]]] = None
    metadata: Dict[str, Any] = field(default_factory=dict)

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

        - ``str`` content is returned verbatim.
        - ``list`` content concatenates the ``text`` of every part whose type
          is ``"text"`` but drops images/audio/video, so consumers (CLI,
          compression, ``AgentResponse.text``) see only readable text.
        """
        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.

  • str content is returned verbatim.
  • list content concatenates the text of every part whose type is "text" but drops images/audio/video, so consumers (CLI, compression, AgentResponse.text) see only readable text.
Source code in src/ciel/runtime/tools.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
def text(self) -> str:
    """Extract plain text from content, tolerant to multimodal parts.

    - ``str`` content is returned verbatim.
    - ``list`` content concatenates the ``text`` of every part whose type
      is ``"text"`` but drops images/audio/video, so consumers (CLI,
      compression, ``AgentResponse.text``) see only readable text.
    """
    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)

ChatRequest dataclass

Source code in src/ciel/runtime/tools.py 
180
181
182
183
184
185
186
187
@dataclass(frozen=True)
class ChatRequest:
    messages: Sequence[ChatMessage]
    tools: Sequence[ToolSpec] = ()
    model: Optional[str] = None
    temperature: Optional[float] = None
    max_tokens: Optional[int] = None
    extra: Dict[str, Any] = field(default_factory=dict)

ChatResponse dataclass

Source code in src/ciel/runtime/tools.py 
190
191
192
193
@dataclass(frozen=True)
class ChatResponse:
    choice: ChatChoice
    metadata: Dict[str, Any] = field(default_factory=dict)

DefaultAgentRuntime

Concrete runtime wiring provider + tool execution with tracing.

Source code in src/ciel/runtime/__init__.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
class DefaultAgentRuntime:
    """Concrete runtime wiring provider + tool execution with tracing."""

    def __init__(
        self,
        *,
        provider: ChatProvider,
        dispatcher: DefaultToolDispatcher,
        registry: Optional[ProviderRegistry] = None,
        audit_sink: Optional[Any] = None,
        agent: str = "default",
        approval_policy: Optional[Any] = None,
    ) -> None:
        self.provider = provider
        self.dispatcher = dispatcher
        self.registry = registry
        self.audit_sink = audit_sink or NullAuditSink()
        self.agent = agent
        self.approval_policy = approval_policy

    async def _emit(self, event: AuditEvent, *, tenant_id: Optional[str] = None) -> AuditEvent:
        normalized = propagate(event, tenant_id=tenant_id)
        await self.audit_sink.write(normalized)
        return normalized

    async def run_agent_loop(
        self,
        *,
        request: ChatRequest,
        tenant_id: Optional[str] = None,
        toolset: Optional[str] = None,
        limit: int = 32,
    ) -> AgentRuntimeResult:
        """Run the agentic tool loop (multi-turn ReAct when tools are present).

        The loop issues up to ``limit`` model completions. After each completion
        that requests tool calls, the requested tools are dispatched and their
        results are appended as messages; the model is then called again. The
        loop stops when (a) the model returns no tool calls (``finish_reason``
        ``stop``), (b) ``limit`` turns are exhausted, or (c) there are no tools
        registered in the request.

        Backward compatibility: when ``limit <= 1`` or ``request.tools`` is
        empty, exactly one completion is produced and no tool messages are
        appended (the historical single-step behaviour). Every tool result from
        every turn is preserved in ``loop_results`` so the facade can surface
        the full ``tool_results`` list.
        """
        session_id = request.extra.get("session_id") or str(uuid.uuid4())
        await self._emit(
            AuditEvent(event="agent.loop.start", session_id=session_id, agent=self.agent, tenant_id=tenant_id),
            tenant_id=tenant_id,
        )

        messages: List[ChatMessage] = list(request.messages)
        loop_results: List[ToolLoopResult] = []
        finish_reason = "stop"
        has_tools = bool(request.tools)

        # Multi-turn only when tools exist and the caller allows more than one step.
        max_turns = max(1, limit) if (has_tools and limit > 1) else 1

        final_response = None
        for _ in range(max_turns):
            response = await self.provider.complete(
                ChatRequest(
                    messages=tuple(messages),
                    tools=request.tools,
                    model=request.model,
                    temperature=request.temperature,
                    max_tokens=request.max_tokens,
                    extra={**request.extra, "session_id": session_id, "tenant_id": tenant_id},
                )
            )
            final_response = response
            messages.append(response.choice.message)
            tool_calls = _extract_tool_calls(response)

            if not tool_calls:
                # Model is done; stop the loop.
                finish_reason = response.choice.finish_reason or "stop"
                break

            dispatch_results: List[ToolResult] = []
            for call in tool_calls:
                call_arguments = call.get("arguments") or {}
                decision = None
                if self.approval_policy is not None and call.get("name") not in {None, ""}:
                    try:
                        from ciel.security import ApprovalRequest, ApprovalPolicy
                        if isinstance(self.approval_policy, type):
                            policy = self.approval_policy()
                        else:
                            policy = self.approval_policy
                        if hasattr(policy, "evaluate"):
                            decision = policy.evaluate(
                                ApprovalRequest(
                                    request_id=call.get("id") or call.get("tool_call_id") or str(uuid.uuid4()),
                                    actor=tenant_id or "unknown",
                                    tool=call.get("name", ""),
                                    arguments=call_arguments,
                                    risk="medium",
                                    tenant=tenant_id,
                                )
                            )
                    except Exception:
                        decision = None
                if decision is not None and not decision.approved:
                    dispatch_results.append(
                        ToolResult(
                            id=call.get("id") or call.get("tool_call_id") or str(uuid.uuid4()),
                            name=call.get("name", ""),
                            error=f"ApprovalDenied: {decision.note or 'denied'}",
                            metadata={"tenant_id": tenant_id, "approval_decision": decision.note or "denied"},
                        )
                    )
                    continue
                dispatch_results.append(
                    await self.dispatcher.dispatch(
                        tenant_id=tenant_id,
                        toolset=toolset or self.dispatcher.default_toolset or "default",
                        name=call.get("name", ""),
                        arguments=call_arguments,
                        tool_call_id=call.get("id") or call.get("tool_call_id") or str(uuid.uuid4()),
                    )
                )

            tool_turn = ToolLoopResult(
                turn_id=str(uuid.uuid4()),
                messages=tuple(messages),
                tool_results=tuple(dispatch_results),
                finish_reason="tool_calls",
                tenant_id=tenant_id,
                metadata={"session_id": session_id, "toolset": toolset, "turn": len(loop_results) + 1},
            )
            loop_results.append(tool_turn)
            finish_reason = "tool_calls"
            await self._emit(
                AuditEvent(
                    event="agent.tool_calls.dispatched",
                    session_id=session_id,
                    agent=self.agent,
                    tool_call_id=dispatch_results[0].id if dispatch_results else None,
                    data={"tools": [r.name for r in dispatch_results], "turn": len(loop_results)},
                    tenant_id=tenant_id,
                ),
                tenant_id=tenant_id,
            )

            # Append tool results as tool messages so the next completion sees them.
            for res in dispatch_results:
                messages.append(
                    ChatMessage(
                        role="tool",
                        content=str(res.output) if res.error is None else (res.error or ""),
                        name=res.name,
                        tool_call_id=res.id,
                        metadata={"tenant_id": tenant_id, "tool_result": True, "error": res.error},
                    )
                )

        await self._emit(
            AuditEvent(event="agent.loop.end", session_id=session_id, agent=self.agent, tenant_id=tenant_id),
            tenant_id=tenant_id,
        )

        return AgentRuntimeResult(
            response=final_response,
            loop_results=tuple(loop_results),
            tenant_id=tenant_id,
            metadata={"session_id": session_id, "agent": self.agent},
        )

    async def stream_agent_loop(
        self,
        *,
        request: ChatRequest,
        tenant_id: Optional[str] = None,
        toolset: Optional[str] = None,
        limit: int = 32,
    ) -> AsyncIterator[ToolLoopResult]:
        result = await self.run_agent_loop(
            request=request,
            tenant_id=tenant_id,
            toolset=toolset,
            limit=limit,
        )
        for turn in result.loop_results:
            yield turn

    async def stream_tokens(
        self,
        *,
        request: ChatRequest,
        tenant_id: Optional[str] = None,
        toolset: Optional[str] = None,
    ) -> AsyncIterator[str]:
        """Stream incremental assistant tokens from the provider.

        Calls ``provider.stream`` (real SSE streaming) and re-emits the
        partial ``content`` of each incremental :class:`ChatResponse` as it
        arrives, so callers see the answer grow token by token.
        """
        chunks = await self.provider.stream(request=request)
        prior = ""
        for chunk in chunks:
            content = chunk.choice.message.text()
            if content != prior:
                yield content
                prior = content

run_agent_loop(*, request: ChatRequest, tenant_id: Optional[str] = None, toolset: Optional[str] = None, limit: int = 32) -> AgentRuntimeResult async

Run the agentic tool loop (multi-turn ReAct when tools are present).

The loop issues up to limit model completions. After each completion that requests tool calls, the requested tools are dispatched and their results are appended as messages; the model is then called again. The loop stops when (a) the model returns no tool calls (finish_reason stop), (b) limit turns are exhausted, or (c) there are no tools registered in the request.

Backward compatibility: when limit <= 1 or request.tools is empty, exactly one completion is produced and no tool messages are appended (the historical single-step behaviour). Every tool result from every turn is preserved in loop_results so the facade can surface the full tool_results list.

Source code in src/ciel/runtime/__init__.py 
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
async def run_agent_loop(
    self,
    *,
    request: ChatRequest,
    tenant_id: Optional[str] = None,
    toolset: Optional[str] = None,
    limit: int = 32,
) -> AgentRuntimeResult:
    """Run the agentic tool loop (multi-turn ReAct when tools are present).

    The loop issues up to ``limit`` model completions. After each completion
    that requests tool calls, the requested tools are dispatched and their
    results are appended as messages; the model is then called again. The
    loop stops when (a) the model returns no tool calls (``finish_reason``
    ``stop``), (b) ``limit`` turns are exhausted, or (c) there are no tools
    registered in the request.

    Backward compatibility: when ``limit <= 1`` or ``request.tools`` is
    empty, exactly one completion is produced and no tool messages are
    appended (the historical single-step behaviour). Every tool result from
    every turn is preserved in ``loop_results`` so the facade can surface
    the full ``tool_results`` list.
    """
    session_id = request.extra.get("session_id") or str(uuid.uuid4())
    await self._emit(
        AuditEvent(event="agent.loop.start", session_id=session_id, agent=self.agent, tenant_id=tenant_id),
        tenant_id=tenant_id,
    )

    messages: List[ChatMessage] = list(request.messages)
    loop_results: List[ToolLoopResult] = []
    finish_reason = "stop"
    has_tools = bool(request.tools)

    # Multi-turn only when tools exist and the caller allows more than one step.
    max_turns = max(1, limit) if (has_tools and limit > 1) else 1

    final_response = None
    for _ in range(max_turns):
        response = await self.provider.complete(
            ChatRequest(
                messages=tuple(messages),
                tools=request.tools,
                model=request.model,
                temperature=request.temperature,
                max_tokens=request.max_tokens,
                extra={**request.extra, "session_id": session_id, "tenant_id": tenant_id},
            )
        )
        final_response = response
        messages.append(response.choice.message)
        tool_calls = _extract_tool_calls(response)

        if not tool_calls:
            # Model is done; stop the loop.
            finish_reason = response.choice.finish_reason or "stop"
            break

        dispatch_results: List[ToolResult] = []
        for call in tool_calls:
            call_arguments = call.get("arguments") or {}
            decision = None
            if self.approval_policy is not None and call.get("name") not in {None, ""}:
                try:
                    from ciel.security import ApprovalRequest, ApprovalPolicy
                    if isinstance(self.approval_policy, type):
                        policy = self.approval_policy()
                    else:
                        policy = self.approval_policy
                    if hasattr(policy, "evaluate"):
                        decision = policy.evaluate(
                            ApprovalRequest(
                                request_id=call.get("id") or call.get("tool_call_id") or str(uuid.uuid4()),
                                actor=tenant_id or "unknown",
                                tool=call.get("name", ""),
                                arguments=call_arguments,
                                risk="medium",
                                tenant=tenant_id,
                            )
                        )
                except Exception:
                    decision = None
            if decision is not None and not decision.approved:
                dispatch_results.append(
                    ToolResult(
                        id=call.get("id") or call.get("tool_call_id") or str(uuid.uuid4()),
                        name=call.get("name", ""),
                        error=f"ApprovalDenied: {decision.note or 'denied'}",
                        metadata={"tenant_id": tenant_id, "approval_decision": decision.note or "denied"},
                    )
                )
                continue
            dispatch_results.append(
                await self.dispatcher.dispatch(
                    tenant_id=tenant_id,
                    toolset=toolset or self.dispatcher.default_toolset or "default",
                    name=call.get("name", ""),
                    arguments=call_arguments,
                    tool_call_id=call.get("id") or call.get("tool_call_id") or str(uuid.uuid4()),
                )
            )

        tool_turn = ToolLoopResult(
            turn_id=str(uuid.uuid4()),
            messages=tuple(messages),
            tool_results=tuple(dispatch_results),
            finish_reason="tool_calls",
            tenant_id=tenant_id,
            metadata={"session_id": session_id, "toolset": toolset, "turn": len(loop_results) + 1},
        )
        loop_results.append(tool_turn)
        finish_reason = "tool_calls"
        await self._emit(
            AuditEvent(
                event="agent.tool_calls.dispatched",
                session_id=session_id,
                agent=self.agent,
                tool_call_id=dispatch_results[0].id if dispatch_results else None,
                data={"tools": [r.name for r in dispatch_results], "turn": len(loop_results)},
                tenant_id=tenant_id,
            ),
            tenant_id=tenant_id,
        )

        # Append tool results as tool messages so the next completion sees them.
        for res in dispatch_results:
            messages.append(
                ChatMessage(
                    role="tool",
                    content=str(res.output) if res.error is None else (res.error or ""),
                    name=res.name,
                    tool_call_id=res.id,
                    metadata={"tenant_id": tenant_id, "tool_result": True, "error": res.error},
                )
            )

    await self._emit(
        AuditEvent(event="agent.loop.end", session_id=session_id, agent=self.agent, tenant_id=tenant_id),
        tenant_id=tenant_id,
    )

    return AgentRuntimeResult(
        response=final_response,
        loop_results=tuple(loop_results),
        tenant_id=tenant_id,
        metadata={"session_id": session_id, "agent": self.agent},
    )

stream_tokens(*, request: ChatRequest, tenant_id: Optional[str] = None, toolset: Optional[str] = None) -> AsyncIterator[str] async

Stream incremental assistant tokens from the provider.

Calls provider.stream (real SSE streaming) and re-emits the partial content of each incremental :class:ChatResponse as it arrives, so callers see the answer grow token by token.

Source code in src/ciel/runtime/__init__.py 
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
async def stream_tokens(
    self,
    *,
    request: ChatRequest,
    tenant_id: Optional[str] = None,
    toolset: Optional[str] = None,
) -> AsyncIterator[str]:
    """Stream incremental assistant tokens from the provider.

    Calls ``provider.stream`` (real SSE streaming) and re-emits the
    partial ``content`` of each incremental :class:`ChatResponse` as it
    arrives, so callers see the answer grow token by token.
    """
    chunks = await self.provider.stream(request=request)
    prior = ""
    for chunk in chunks:
        content = chunk.choice.message.text()
        if content != prior:
            yield content
            prior = content

DefaultToolDispatcher

Dispatch tool requests to a configured ToolProvider.

Source code in src/ciel/runtime/__init__.py 
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
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
class DefaultToolDispatcher:
    """Dispatch tool requests to a configured ToolProvider."""

    provider: ToolProvider
    default_toolset: Optional[str]

    def __init__(self, provider: ToolProvider, default_toolset: Optional[str] = None) -> None:
        self.provider = provider
        self.default_toolset = default_toolset or getattr(provider.registry, "default_toolset", None)

    async def dispatch(
        self,
        *,
        tenant_id: Optional[str] = None,
        toolset: Optional[str] = None,
        name: str,
        arguments: Dict[str, Any],
        tool_call_id: str,
    ) -> ToolResult:
        result = await self.provider.execute(
            tenant_id=tenant_id,
            toolset=toolset or self.default_toolset or "default",
            name=name,
            arguments=arguments,
            tool_call_id=tool_call_id,
        )
        result.metadata.setdefault("tenant_id", tenant_id)
        return result

    async def dispatch_all(
        self,
        *,
        tenant_id: Optional[str] = None,
        toolset: Optional[str] = None,
        calls: Sequence[Dict[str, Any]],
    ):
        results: List[ToolResult] = []
        for call in calls:
            call_tenant_id = tenant_id or call.get("metadata", {}).get("tenant_id")
            results.append(
                await self.dispatch(
                    tenant_id=call_tenant_id,
                    toolset=toolset or self.default_toolset or "default",
                    name=call["name"],
                    arguments=call.get("arguments", {}),
                    tool_call_id=call.get("id") or call.get("tool_call_id") or str(uuid.uuid4()),
                )
            )
        return results

HealthResponse

Bases: BaseModel

Source code in src/ciel/gateway/base.py 
30
31
32
class HealthResponse(BaseModel):
    status: str = "ok"
    version: str

InfoResponse

Bases: BaseModel

Source code in src/ciel/gateway/base.py 
57
58
59
60
class InfoResponse(BaseModel):
    version: str
    providers: List[ProviderInfo]
    default_tenant: Optional[str] = None

KanbanBoard

In-memory kanban board, optionally backed by a durable SQLite store.

When path is None (the default) the board keeps tasks in a process local dict exactly as before -- this preserves the existing in-memory API used by the gateway and the CLI. When path is provided the board transparently reads and writes every task to a SQLite database opened in WAL mode, so state survives process restarts.

Source code in src/ciel/orchestration/board.py 
 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
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 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
128
129
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
181
182
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
class KanbanBoard:
    """In-memory kanban board, optionally backed by a durable SQLite store.

    When ``path`` is ``None`` (the default) the board keeps tasks in a process
    local ``dict`` exactly as before -- this preserves the existing in-memory
    API used by the gateway and the CLI. When ``path`` is provided the board
    transparently reads and writes every task to a SQLite database opened in
    WAL mode, so state survives process restarts.
    """

    def __init__(self, path: Optional[str | Path] = None) -> None:
        self._path = None if path is None else str(path)
        self._tasks: Dict[str, BoardTask] = {}
        self._conn: Optional[sqlite3.Connection] = None
        if self._path is not None:
            self._conn = sqlite3.connect(self._path)
            self._conn.execute("PRAGMA journal_mode=WAL")
            self._conn.execute(
                """
                CREATE TABLE IF NOT EXISTS board_tasks (
                    id TEXT PRIMARY KEY,
                    title TEXT NOT NULL,
                    status TEXT NOT NULL,
                    assignee TEXT,
                    tenant_id TEXT,
                    metadata_json TEXT
                )
                """
            )
            self._conn.commit()

    # -- helpers -------------------------------------------------------------
    def _row_to_task(self, row: sqlite3.Row) -> BoardTask:
        return BoardTask(
            id=row["id"],
            title=row["title"],
            status=row["status"],
            assignee=row["assignee"],
            tenant_id=row["tenant_id"],
            metadata=json.loads(row["metadata_json"]) if row["metadata_json"] else {},
        )

    def _task_to_row(self, task: BoardTask) -> Dict[str, Any]:
        return {
            "id": task.id,
            "title": task.title,
            "status": task.status,
            "assignee": task.assignee,
            "tenant_id": task.tenant_id,
            "metadata_json": json.dumps(task.metadata or {}),
        }

    # -- public API ----------------------------------------------------------
    def add_task(self, task: BoardTask) -> BoardTask:
        if self._conn is None:
            self._tasks[task.id] = task
            return task

        row = self._task_to_row(task)
        self._conn.execute(
            """
            INSERT INTO board_tasks (id, title, status, assignee, tenant_id, metadata_json)
            VALUES (:id, :title, :status, :assignee, :tenant_id, :metadata_json)
            ON CONFLICT(id) DO UPDATE SET
                title=excluded.title,
                status=excluded.status,
                assignee=excluded.assignee,
                tenant_id=excluded.tenant_id,
                metadata_json=excluded.metadata_json
            """,
            row,
        )
        self._conn.commit()
        return task

    def assign(self, task_id: str, assignee: str) -> Optional[BoardTask]:
        if self._conn is None:
            task = self._tasks.get(task_id)
            if not task:
                return None
            task.assignee = assignee
            task.status = "in_progress"
            return task

        cur = self._conn.execute(
            """
            UPDATE board_tasks
            SET assignee = ?, status = 'in_progress'
            WHERE id = ?
            """,
            (assignee, task_id),
        )
        if cur.rowcount == 0:
            return None
        self._conn.commit()
        return self.show(task_id)

    def move(self, task_id: str, status: str) -> Optional[BoardTask]:
        if self._conn is None:
            task = self._tasks.get(task_id)
            if not task:
                return None
            task.status = status
            return task

        cur = self._conn.execute(
            "UPDATE board_tasks SET status = ? WHERE id = ?",
            (status, task_id),
        )
        if cur.rowcount == 0:
            return None
        self._conn.commit()
        return self.show(task_id)

    def list_tasks(
        self,
        status: Optional[str] = None,
        assignee: Optional[str] = None,
        tenant_id: Optional[str] = None,
    ) -> List[BoardTask]:
        if self._conn is None:
            out: List[BoardTask] = []
            for task in self._tasks.values():
                if status and task.status != status:
                    continue
                if assignee and task.assignee != assignee:
                    continue
                if tenant_id and task.tenant_id != tenant_id:
                    continue
                out.append(task)
            return out

        clauses: List[str] = []
        params: List[Any] = []
        if status:
            clauses.append("status = ?")
            params.append(status)
        if assignee:
            clauses.append("assignee = ?")
            params.append(assignee)
        if tenant_id:
            clauses.append("tenant_id = ?")
            params.append(tenant_id)
        where = (" WHERE " + " AND ".join(clauses)) if clauses else ""
        self._conn.row_factory = sqlite3.Row
        rows = self._conn.execute(
            f"SELECT id, title, status, assignee, tenant_id, metadata_json FROM board_tasks{where}",
            params,
        ).fetchall()
        return [self._row_to_task(r) for r in rows]

    def show(self, task_id: str) -> Optional[BoardTask]:
        if self._conn is None:
            return self._tasks.get(task_id)

        self._conn.row_factory = sqlite3.Row
        row = self._conn.execute(
            "SELECT id, title, status, assignee, tenant_id, metadata_json FROM board_tasks WHERE id = ?",
            (task_id,),
        ).fetchone()
        if row is None:
            return None
        return self._row_to_task(row)

    def close(self) -> None:
        if self._conn is not None:
            # Flush the WAL back into the main database file so the
            # board.sqlite-wal / board.sqlite-shm side-car files are released
            # (important on Windows where they otherwise stay locked).
            try:
                self._conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
            except sqlite3.Error:
                pass
            self._conn.close()
            self._conn = None

    def __enter__(self) -> "KanbanBoard":
        return self

    def __exit__(self, *exc: Any) -> None:
        self.close()

LivenessResponse

Bases: BaseModel

Liveness: el proceso está vivo (no depende de backends remotos).

Source code in src/ciel/gateway/base.py 
35
36
37
38
39
class LivenessResponse(BaseModel):
    """Liveness: el proceso está vivo (no depende de backends remotos)."""

    status: str = "alive"
    version: str

ProviderInfo

Bases: BaseModel

Source code in src/ciel/gateway/base.py 
52
53
54
class ProviderInfo(BaseModel):
    name: str
    provider_name: Optional[str] = None

ProviderRegistry

Source code in src/ciel/providers/__init__.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
class ProviderRegistry:
    def __init__(self) -> None:
        self._providers: Dict[str, ChatProvider] = {}
        self._configs: Dict[str, Dict[str, Any]] = {}

    def register(self, name: str, provider: ChatProvider, *, config: Optional[Dict[str, Any]] = None) -> None:
        self._providers[name] = provider
        self._configs[name] = config or {}

    def get(self, name: str) -> ChatProvider:
        if name not in self._providers:
            raise _domain_error(f"Provider not registered: {name}")
        return self._providers[name]

    def available(self) -> Sequence[str]:
        return list(self._providers.keys())

ReadinessResponse

Bases: BaseModel

Readiness: el gateway puede atender tráfico (backend conectado + migrado).

Source code in src/ciel/gateway/base.py 
42
43
44
45
46
47
48
49
class ReadinessResponse(BaseModel):
    """Readiness: el gateway puede atender tráfico (backend conectado + migrado)."""

    status: str  # "ready" | "not_ready"
    version: str
    backend: str
    backend_ready: bool
    details: Dict[str, Any] = Field(default_factory=dict)

ToolInvokeRequest

Bases: BaseModel

Source code in src/ciel/gateway/base.py 
88
89
90
91
class ToolInvokeRequest(BaseModel):
    arguments: Dict[str, Any] = Field(default_factory=dict)
    tool_call_id: Optional[str] = None
    tenant_id: Optional[str] = None

ToolInvokeResponse

Bases: BaseModel

Source code in src/ciel/gateway/base.py 
94
95
96
97
98
99
class ToolInvokeResponse(BaseModel):
    id: Optional[str] = None
    name: str
    output: Any = None
    error: Optional[str] = None
    metadata: Dict[str, Any] = Field(default_factory=dict)

ToolResult dataclass

Source code in src/ciel/runtime/tools.py 
25
26
27
28
29
30
31
32
33
@dataclass
class ToolResult:
    id: str
    name: str
    output: Any = None
    error: Optional[str] = None
    usage: Optional[Dict[str, Any]] = None
    duration_ms: Optional[int] = None
    metadata: Dict[str, Any] = field(default_factory=dict)

ToolResultPayload

Bases: BaseModel

Source code in src/ciel/gateway/base.py 
74
75
76
77
78
79
class ToolResultPayload(BaseModel):
    id: Optional[str] = None
    name: str
    output: Any = None
    error: Optional[str] = None
    metadata: Dict[str, Any] = Field(default_factory=dict)

_list_board_tasks(board: KanbanBoard, tenant_id: Optional[str]) -> List[BoardTaskPayload]

Source code in src/ciel/gateway/base.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
def _list_board_tasks(board: KanbanBoard, tenant_id: Optional[str]) -> List[BoardTaskPayload]:
    tasks = board.list_tasks(tenant_id=tenant_id)
    out: List[BoardTaskPayload] = []
    for task in tasks:
        out.append(
            BoardTaskPayload(
                id=task.id,
                title=task.title,
                status=task.status,
                assignee=task.assignee,
                tenant_id=task.tenant_id,
                metadata=dict(task.metadata or {}),
            )
        )
    return out

_serialize_chat_response(result: AgentRuntimeResult) -> AgentRunResponse

Source code in src/ciel/gateway/base.py 
127
128
129
130
131
132
133
134
135
def _serialize_chat_response(result: AgentRuntimeResult) -> AgentRunResponse:
    response: ChatResponse = result.response
    text = getattr(response.choice.message, "content", "") or ""
    session_id = result.metadata.get("session_id")
    tool_results: List[ToolResultPayload] = []
    for turn in result.loop_results:
        for tr in turn.tool_results:
            tool_results.append(_tool_result_to_payload(tr))
    return AgentRunResponse(text=text, session_id=session_id, tool_results=tool_results)

_tool_result_to_payload(result: ToolResult) -> ToolResultPayload

Source code in src/ciel/gateway/base.py 
117
118
119
120
121
122
123
124
def _tool_result_to_payload(result: ToolResult) -> ToolResultPayload:
    return ToolResultPayload(
        id=getattr(result, "id", None),
        name=getattr(result, "name", ""),
        output=getattr(result, "output", None),
        error=getattr(result, "error", None),
        metadata=dict(getattr(result, "metadata", {}) or {}),
    )

create_control_app(*, runtime: DefaultAgentRuntime, registry: Optional[ProviderRegistry] = None, audit_sink: Any = None, tenant_id: Optional[str] = None, api_key: Optional[str] = None, board_db_path: Optional[str] = None) -> FastAPI

Build a FastAPI control plane for the Ciel agent runtime.

Parameters

runtime: A :class:ciel.runtime.DefaultAgentRuntime used to run agent loops and dispatch tools. Tool dispatch goes through runtime.dispatcher. El endpoint POST /v1/agent/run/stream usa runtime.stream_tokens para emitir Server-Sent Events (SSE) con los fragmentos incrementales del assistant y cierra con data: [DONE]. registry: Optional :class:ciel.providers.ProviderRegistry surfaced via the GET /info endpoint. If omitted, an empty provider list is reported. audit_sink: Optional audit sink retained for observability hooks. Not directly invoked here — the runtime already emits audit events. tenant_id: Default tenant propagated to the runtime when a request does not provide its own tenant_id. api_key: Optional transport API key. When None (the default), the CIEL_API_KEY environment variable is consulted. If a key is configured (explicitly or via env), protected routes require a valid key via Authorization: Bearer *** orX-API-Key; otherwise the dependency is a no-op (open mode). board_db_path: Optional path to a SQLite database for the kanban board. When set (o bien vía la variable de entornoCIEL_BOARD_DB), el board se monta sobre SQLite para que/v1/board/list`` vea las tareas creadas por el CLI y viceversa. Si no se configura ninguna ruta, el board queda en memoria (comportamiento legacy, útil para smoke tests/offline).

Source code in src/ciel/gateway/base.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
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
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
def create_control_app(
    *,
    runtime: DefaultAgentRuntime,
    registry: Optional[ProviderRegistry] = None,
    audit_sink: Any = None,
    tenant_id: Optional[str] = None,
    api_key: Optional[str] = None,
    board_db_path: Optional[str] = None,
) -> FastAPI:
    """Build a FastAPI control plane for the Ciel agent runtime.

    Parameters
    ----------
    runtime:
        A :class:`ciel.runtime.DefaultAgentRuntime` used to run agent loops and
        dispatch tools. Tool dispatch goes through ``runtime.dispatcher``.
        El endpoint ``POST /v1/agent/run/stream`` usa
        ``runtime.stream_tokens`` para emitir Server-Sent Events (SSE) con los
        fragmentos incrementales del assistant y cierra con ``data: [DONE]``.
    registry:
        Optional :class:`ciel.providers.ProviderRegistry` surfaced via the
        ``GET /info`` endpoint. If omitted, an empty provider list is reported.
    audit_sink:
        Optional audit sink retained for observability hooks. Not directly
        invoked here — the runtime already emits audit events.
    tenant_id:
        Default tenant propagated to the runtime when a request does not
        provide its own ``tenant_id``.
    api_key:
        Optional transport API key. When ``None`` (the default), the
        ``CIEL_API_KEY`` environment variable is consulted. If a key is
        configured (explicitly or via env), protected routes require a valid
        key via ``Authorization: Bearer *** or ``X-API-Key``; otherwise
        the dependency is a no-op (open mode).
    board_db_path:
        Optional path to a SQLite database for the kanban board. When set (o
        bien vía la variable de entorno ``CIEL_BOARD_DB``), el board se monta
        sobre SQLite para que ``/v1/board/list`` vea las tareas creadas por el
        CLI y viceversa. Si no se configura ninguna ruta, el board queda en
        memoria (comportamiento legacy, útil para smoke tests/offline).
    """
    from ciel.gateway.auth import Depends, make_auth_dependency

    api_key_guard = Depends(make_auth_dependency(expected_key=api_key))

    app = FastAPI(
        title="Ciel Control Gateway",
        version=__version__,
        description="HTTP control plane for the Ciel agent runtime.",
    )
    # Attach references for tests / introspection.
    app.state.runtime = runtime
    app.state.registry = registry
    app.state.audit_sink = audit_sink
    app.state.default_tenant_id = tenant_id

    # Resolve a live board instance if orchestration is available.
    # Si se pasa board_db_path o existe CIEL_BOARD_DB, el board se monta sobre
    # SQLite para compartir estado con el CLI; en caso contrario queda en
    # memoria (fallback para smoke tests/offline).
    board: Optional[KanbanBoard] = None
    resolved_board_db = board_db_path or os.environ.get("CIEL_BOARD_DB")
    try:
        board = KanbanBoard(path=resolved_board_db)  # None => en memoria.
    except Exception:  # pragma: no cover - defensive guard
        board = None
    app.state.board = board

    def _resolve_tenant(request_tenant: Optional[str]) -> Optional[str]:
        return request_tenant or tenant_id

    @app.get("/health", response_model=HealthResponse, tags=["meta"])
    async def health() -> HealthResponse:
        return HealthResponse(version=__version__)

    # --- Fase 14 / F16: health reales (liveness vs readiness) ---------------
    @app.get("/healthz", response_model=LivenessResponse, tags=["meta"])
    async def healthz() -> LivenessResponse:
        # Liveness: el proceso está vivo. No depende de backends remotos.
        return LivenessResponse(status="alive", version=__version__)

    @app.get("/readyz", response_model=ReadinessResponse, tags=["meta"])
    async def readyz(request: Request) -> ReadinessResponse:
        # Readiness: el gateway puede atender tráfico. Depende de que el
        # StateBackend compartido esté conectado y migrado.
        backend = getattr(request.app.state, "state_backend", None)
        if backend is None:
            # Sin backend cableado => no listo para multi-réplica.
            return ReadinessResponse(
                status="not_ready",
                version=__version__,
                backend="none",
                backend_ready=False,
                details={"reason": "state_backend not wired"},
            )
        ready = backend.is_ready()
        return ReadinessResponse(
            status="ready" if ready else "not_ready",
            version=__version__,
            backend=backend.backend_type,
            backend_ready=ready,
            details={"ready": ready},
        )

    @app.get("/info", response_model=InfoResponse, tags=["meta"])
    async def info() -> InfoResponse:
        providers: List[ProviderInfo] = []
        if registry is not None:
            for name in registry.available():
                provider_name: Optional[str] = None
                try:
                    provider = registry.get(name)
                    provider_name = getattr(provider, "provider_name", None)
                except Exception:
                    provider_name = None
                providers.append(ProviderInfo(name=name, provider_name=provider_name))
        return InfoResponse(
            version=__version__,
            providers=providers,
            default_tenant=tenant_id,
        )

    @app.post(
        "/v1/agent/run",
        response_model=AgentRunResponse,
        tags=["agent"],
        dependencies=[api_key_guard],
    )
    async def agent_run(body: AgentRunRequest) -> AgentRunResponse:
        effective_tenant = _resolve_tenant(body.tenant_id)
        if effective_tenant is None:
            raise HTTPException(
                status_code=400,
                detail="tenant_id is required (multi-tenancy is enforced). "
                "Provide it in the request or configure a default tenant in the gateway.",
            )
        session_id = body.session_id or str(uuid.uuid4())
        request = _build_chat_request(body, session_id, effective_tenant)
        try:
            result = await runtime.run_agent_loop(
                request=request,
                tenant_id=effective_tenant,
                toolset=body.toolset,
            )
        except Exception as exc:
            logger.exception("agent.run failed: %s", exc)
            raise HTTPException(status_code=500, detail=f"agent loop failed: {exc}") from exc
        return _serialize_chat_response(result)

    def _build_chat_request(body: AgentRunRequest, session_id: str, effective_tenant: Optional[str]) -> ChatRequest:
        extra: Dict[str, Any] = dict(body.extra or {})
        extra.setdefault("session_id", session_id)
        if effective_tenant is not None:
            extra.setdefault("tenant_id", effective_tenant)
        return ChatRequest(
            messages=(ChatMessage(role="user", content=body.prompt),),
            tools=(),
            model=body.model,
            temperature=body.temperature,
            max_tokens=body.max_tokens,
            extra=extra,
        )

    @app.post(
        "/v1/agent/run/stream",
        tags=["agent"],
        dependencies=[api_key_guard],
    )
    async def agent_run_stream(body: AgentRunRequest) -> StreamingResponse:
        """Stream the assistant completion as Server-Sent Events.

        Emite cada fragmento incremental del assistant como un evento SSE
        ``data: <token>`` y cierra la respuesta con ``data: [DONE]``.
        Reutiliza ``runtime.stream_tokens`` (que a su vez llama a
        ``provider.stream`` para hacer SSE real cuando el proveedor lo soporta).

        El ``tenant_id`` es obligatorio, igual que en ``/v1/agent/run``.
        """
        effective_tenant = _resolve_tenant(body.tenant_id)
        if effective_tenant is None:
            raise HTTPException(
                status_code=400,
                detail="tenant_id is required (multi-tenancy is enforced). "
                "Provide it in the request or configure a default tenant in the gateway.",
            )
        session_id = body.session_id or str(uuid.uuid4())

        async def _event_generator():
            try:
                request = _build_chat_request(body, session_id, effective_tenant)
                async for token in runtime.stream_tokens(
                    request=request,
                    tenant_id=effective_tenant,
                    toolset=body.toolset,
                ):
                    yield f"data: {token}\n\n"
            except Exception as exc:  # pragma: no cover - defensive guard
                logger.exception("agent.run.stream failed: %s", exc)
                yield f"data: [ERROR] {exc}\n\n"
            yield "data: [DONE]\n\n"

        return StreamingResponse(_event_generator(), media_type="text/event-stream")

    @app.post(
        "/v1/tools/{toolset}/{name}",
        response_model=ToolInvokeResponse,
        tags=["tools"],
        dependencies=[api_key_guard],
    )
    async def tool_invoke(toolset: str, name: str, body: ToolInvokeRequest) -> ToolInvokeResponse:
        dispatcher: Optional[DefaultToolDispatcher] = getattr(runtime, "dispatcher", None)
        if dispatcher is None:
            raise HTTPException(status_code=500, detail="runtime has no tool dispatcher")
        effective_tenant = _resolve_tenant(body.tenant_id)
        if effective_tenant is None:
            raise HTTPException(
                status_code=400,
                detail="tenant_id is required to dispatch tools (multi-tenancy is enforced).",
            )
        tool_call_id = body.tool_call_id or str(uuid.uuid4())
        try:
            result = await dispatcher.dispatch(
                tenant_id=effective_tenant,
                toolset=toolset,
                name=name,
                arguments=dict(body.arguments or {}),
                tool_call_id=tool_call_id,
            )
        except Exception as exc:
            logger.exception("tool.invoke failed: %s", exc)
            raise HTTPException(status_code=500, detail=f"tool dispatch failed: {exc}") from exc
        return ToolInvokeResponse(
            id=getattr(result, "id", None),
            name=getattr(result, "name", name),
            output=getattr(result, "output", None),
            error=getattr(result, "error", None),
            metadata=dict(getattr(result, "metadata", {}) or {}),
        )

    @app.get(
        "/v1/board/list",
        response_model=BoardListResponse,
        tags=["board"],
        dependencies=[api_key_guard],
    )
    async def board_list(tenant_id: Optional[str] = None) -> BoardListResponse:
        effective_tenant = _resolve_tenant(tenant_id)
        if board is None:
            return BoardListResponse(tasks=[])
        return BoardListResponse(tasks=_list_board_tasks(board, effective_tenant))

    return app

Composed Ciel gateway application (control + MCP host + webhook).

:func:make_app builds a single FastAPI application that exposes the three verified gateway surfaces on one port:

  • control plane -> :func:ciel.gateway.base.create_control_app (/health, /info, /v1/agent/run, /v1/tools/..., /v1/board/list)
  • MCP host -> :func:ciel.gateway.mount_mcp_app mounted at /mcp (JSON-RPC POST /mcp/)
  • webhook -> :func:ciel.gateway.create_webhook_router at /v1/messaging/webhook (POST + /health)

The runtime is built from environment configuration so ciel serve can start with zero external dependencies: when no CIEL_PROVIDER_URL is set a local echo provider is used, which keeps the gateway bootable for smoke tests and offline deployments. Multi-tenancy is never relaxed — the control plane still requires tenant_id on runtime/tool requests.

__all__ = ['make_app', '_EchoProvider'] module-attribute

__version__ = _pkg_version('ciel') module-attribute

logger = logging.getLogger(__name__) module-attribute

ChatChoice dataclass

Source code in src/ciel/runtime/tools.py 
172
173
174
175
176
177
@dataclass(frozen=True)
class ChatChoice:
    message: ChatMessage
    finish_reason: str
    usage: Optional[Dict[str, Any]] = None
    metadata: Dict[str, Any] = field(default_factory=dict)

ChatMessage dataclass

Source code in src/ciel/runtime/tools.py 
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
@dataclass(frozen=True)
class ChatMessage:
    role: str
    content: ChatContent
    name: Optional[str] = None
    tool_call_id: Optional[str] = None
    tool_calls: Optional[list[dict[str, Any]]] = None
    metadata: Dict[str, Any] = field(default_factory=dict)

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

        - ``str`` content is returned verbatim.
        - ``list`` content concatenates the ``text`` of every part whose type
          is ``"text"`` but drops images/audio/video, so consumers (CLI,
          compression, ``AgentResponse.text``) see only readable text.
        """
        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.

  • str content is returned verbatim.
  • list content concatenates the text of every part whose type is "text" but drops images/audio/video, so consumers (CLI, compression, AgentResponse.text) see only readable text.
Source code in src/ciel/runtime/tools.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
def text(self) -> str:
    """Extract plain text from content, tolerant to multimodal parts.

    - ``str`` content is returned verbatim.
    - ``list`` content concatenates the ``text`` of every part whose type
      is ``"text"`` but drops images/audio/video, so consumers (CLI,
      compression, ``AgentResponse.text``) see only readable text.
    """
    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)

ChatResponse dataclass

Source code in src/ciel/runtime/tools.py 
190
191
192
193
@dataclass(frozen=True)
class ChatResponse:
    choice: ChatChoice
    metadata: Dict[str, Any] = field(default_factory=dict)

DefaultAgentRuntime

Concrete runtime wiring provider + tool execution with tracing.

Source code in src/ciel/runtime/__init__.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
class DefaultAgentRuntime:
    """Concrete runtime wiring provider + tool execution with tracing."""

    def __init__(
        self,
        *,
        provider: ChatProvider,
        dispatcher: DefaultToolDispatcher,
        registry: Optional[ProviderRegistry] = None,
        audit_sink: Optional[Any] = None,
        agent: str = "default",
        approval_policy: Optional[Any] = None,
    ) -> None:
        self.provider = provider
        self.dispatcher = dispatcher
        self.registry = registry
        self.audit_sink = audit_sink or NullAuditSink()
        self.agent = agent
        self.approval_policy = approval_policy

    async def _emit(self, event: AuditEvent, *, tenant_id: Optional[str] = None) -> AuditEvent:
        normalized = propagate(event, tenant_id=tenant_id)
        await self.audit_sink.write(normalized)
        return normalized

    async def run_agent_loop(
        self,
        *,
        request: ChatRequest,
        tenant_id: Optional[str] = None,
        toolset: Optional[str] = None,
        limit: int = 32,
    ) -> AgentRuntimeResult:
        """Run the agentic tool loop (multi-turn ReAct when tools are present).

        The loop issues up to ``limit`` model completions. After each completion
        that requests tool calls, the requested tools are dispatched and their
        results are appended as messages; the model is then called again. The
        loop stops when (a) the model returns no tool calls (``finish_reason``
        ``stop``), (b) ``limit`` turns are exhausted, or (c) there are no tools
        registered in the request.

        Backward compatibility: when ``limit <= 1`` or ``request.tools`` is
        empty, exactly one completion is produced and no tool messages are
        appended (the historical single-step behaviour). Every tool result from
        every turn is preserved in ``loop_results`` so the facade can surface
        the full ``tool_results`` list.
        """
        session_id = request.extra.get("session_id") or str(uuid.uuid4())
        await self._emit(
            AuditEvent(event="agent.loop.start", session_id=session_id, agent=self.agent, tenant_id=tenant_id),
            tenant_id=tenant_id,
        )

        messages: List[ChatMessage] = list(request.messages)
        loop_results: List[ToolLoopResult] = []
        finish_reason = "stop"
        has_tools = bool(request.tools)

        # Multi-turn only when tools exist and the caller allows more than one step.
        max_turns = max(1, limit) if (has_tools and limit > 1) else 1

        final_response = None
        for _ in range(max_turns):
            response = await self.provider.complete(
                ChatRequest(
                    messages=tuple(messages),
                    tools=request.tools,
                    model=request.model,
                    temperature=request.temperature,
                    max_tokens=request.max_tokens,
                    extra={**request.extra, "session_id": session_id, "tenant_id": tenant_id},
                )
            )
            final_response = response
            messages.append(response.choice.message)
            tool_calls = _extract_tool_calls(response)

            if not tool_calls:
                # Model is done; stop the loop.
                finish_reason = response.choice.finish_reason or "stop"
                break

            dispatch_results: List[ToolResult] = []
            for call in tool_calls:
                call_arguments = call.get("arguments") or {}
                decision = None
                if self.approval_policy is not None and call.get("name") not in {None, ""}:
                    try:
                        from ciel.security import ApprovalRequest, ApprovalPolicy
                        if isinstance(self.approval_policy, type):
                            policy = self.approval_policy()
                        else:
                            policy = self.approval_policy
                        if hasattr(policy, "evaluate"):
                            decision = policy.evaluate(
                                ApprovalRequest(
                                    request_id=call.get("id") or call.get("tool_call_id") or str(uuid.uuid4()),
                                    actor=tenant_id or "unknown",
                                    tool=call.get("name", ""),
                                    arguments=call_arguments,
                                    risk="medium",
                                    tenant=tenant_id,
                                )
                            )
                    except Exception:
                        decision = None
                if decision is not None and not decision.approved:
                    dispatch_results.append(
                        ToolResult(
                            id=call.get("id") or call.get("tool_call_id") or str(uuid.uuid4()),
                            name=call.get("name", ""),
                            error=f"ApprovalDenied: {decision.note or 'denied'}",
                            metadata={"tenant_id": tenant_id, "approval_decision": decision.note or "denied"},
                        )
                    )
                    continue
                dispatch_results.append(
                    await self.dispatcher.dispatch(
                        tenant_id=tenant_id,
                        toolset=toolset or self.dispatcher.default_toolset or "default",
                        name=call.get("name", ""),
                        arguments=call_arguments,
                        tool_call_id=call.get("id") or call.get("tool_call_id") or str(uuid.uuid4()),
                    )
                )

            tool_turn = ToolLoopResult(
                turn_id=str(uuid.uuid4()),
                messages=tuple(messages),
                tool_results=tuple(dispatch_results),
                finish_reason="tool_calls",
                tenant_id=tenant_id,
                metadata={"session_id": session_id, "toolset": toolset, "turn": len(loop_results) + 1},
            )
            loop_results.append(tool_turn)
            finish_reason = "tool_calls"
            await self._emit(
                AuditEvent(
                    event="agent.tool_calls.dispatched",
                    session_id=session_id,
                    agent=self.agent,
                    tool_call_id=dispatch_results[0].id if dispatch_results else None,
                    data={"tools": [r.name for r in dispatch_results], "turn": len(loop_results)},
                    tenant_id=tenant_id,
                ),
                tenant_id=tenant_id,
            )

            # Append tool results as tool messages so the next completion sees them.
            for res in dispatch_results:
                messages.append(
                    ChatMessage(
                        role="tool",
                        content=str(res.output) if res.error is None else (res.error or ""),
                        name=res.name,
                        tool_call_id=res.id,
                        metadata={"tenant_id": tenant_id, "tool_result": True, "error": res.error},
                    )
                )

        await self._emit(
            AuditEvent(event="agent.loop.end", session_id=session_id, agent=self.agent, tenant_id=tenant_id),
            tenant_id=tenant_id,
        )

        return AgentRuntimeResult(
            response=final_response,
            loop_results=tuple(loop_results),
            tenant_id=tenant_id,
            metadata={"session_id": session_id, "agent": self.agent},
        )

    async def stream_agent_loop(
        self,
        *,
        request: ChatRequest,
        tenant_id: Optional[str] = None,
        toolset: Optional[str] = None,
        limit: int = 32,
    ) -> AsyncIterator[ToolLoopResult]:
        result = await self.run_agent_loop(
            request=request,
            tenant_id=tenant_id,
            toolset=toolset,
            limit=limit,
        )
        for turn in result.loop_results:
            yield turn

    async def stream_tokens(
        self,
        *,
        request: ChatRequest,
        tenant_id: Optional[str] = None,
        toolset: Optional[str] = None,
    ) -> AsyncIterator[str]:
        """Stream incremental assistant tokens from the provider.

        Calls ``provider.stream`` (real SSE streaming) and re-emits the
        partial ``content`` of each incremental :class:`ChatResponse` as it
        arrives, so callers see the answer grow token by token.
        """
        chunks = await self.provider.stream(request=request)
        prior = ""
        for chunk in chunks:
            content = chunk.choice.message.text()
            if content != prior:
                yield content
                prior = content

run_agent_loop(*, request: ChatRequest, tenant_id: Optional[str] = None, toolset: Optional[str] = None, limit: int = 32) -> AgentRuntimeResult async

Run the agentic tool loop (multi-turn ReAct when tools are present).

The loop issues up to limit model completions. After each completion that requests tool calls, the requested tools are dispatched and their results are appended as messages; the model is then called again. The loop stops when (a) the model returns no tool calls (finish_reason stop), (b) limit turns are exhausted, or (c) there are no tools registered in the request.

Backward compatibility: when limit <= 1 or request.tools is empty, exactly one completion is produced and no tool messages are appended (the historical single-step behaviour). Every tool result from every turn is preserved in loop_results so the facade can surface the full tool_results list.

Source code in src/ciel/runtime/__init__.py 
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
async def run_agent_loop(
    self,
    *,
    request: ChatRequest,
    tenant_id: Optional[str] = None,
    toolset: Optional[str] = None,
    limit: int = 32,
) -> AgentRuntimeResult:
    """Run the agentic tool loop (multi-turn ReAct when tools are present).

    The loop issues up to ``limit`` model completions. After each completion
    that requests tool calls, the requested tools are dispatched and their
    results are appended as messages; the model is then called again. The
    loop stops when (a) the model returns no tool calls (``finish_reason``
    ``stop``), (b) ``limit`` turns are exhausted, or (c) there are no tools
    registered in the request.

    Backward compatibility: when ``limit <= 1`` or ``request.tools`` is
    empty, exactly one completion is produced and no tool messages are
    appended (the historical single-step behaviour). Every tool result from
    every turn is preserved in ``loop_results`` so the facade can surface
    the full ``tool_results`` list.
    """
    session_id = request.extra.get("session_id") or str(uuid.uuid4())
    await self._emit(
        AuditEvent(event="agent.loop.start", session_id=session_id, agent=self.agent, tenant_id=tenant_id),
        tenant_id=tenant_id,
    )

    messages: List[ChatMessage] = list(request.messages)
    loop_results: List[ToolLoopResult] = []
    finish_reason = "stop"
    has_tools = bool(request.tools)

    # Multi-turn only when tools exist and the caller allows more than one step.
    max_turns = max(1, limit) if (has_tools and limit > 1) else 1

    final_response = None
    for _ in range(max_turns):
        response = await self.provider.complete(
            ChatRequest(
                messages=tuple(messages),
                tools=request.tools,
                model=request.model,
                temperature=request.temperature,
                max_tokens=request.max_tokens,
                extra={**request.extra, "session_id": session_id, "tenant_id": tenant_id},
            )
        )
        final_response = response
        messages.append(response.choice.message)
        tool_calls = _extract_tool_calls(response)

        if not tool_calls:
            # Model is done; stop the loop.
            finish_reason = response.choice.finish_reason or "stop"
            break

        dispatch_results: List[ToolResult] = []
        for call in tool_calls:
            call_arguments = call.get("arguments") or {}
            decision = None
            if self.approval_policy is not None and call.get("name") not in {None, ""}:
                try:
                    from ciel.security import ApprovalRequest, ApprovalPolicy
                    if isinstance(self.approval_policy, type):
                        policy = self.approval_policy()
                    else:
                        policy = self.approval_policy
                    if hasattr(policy, "evaluate"):
                        decision = policy.evaluate(
                            ApprovalRequest(
                                request_id=call.get("id") or call.get("tool_call_id") or str(uuid.uuid4()),
                                actor=tenant_id or "unknown",
                                tool=call.get("name", ""),
                                arguments=call_arguments,
                                risk="medium",
                                tenant=tenant_id,
                            )
                        )
                except Exception:
                    decision = None
            if decision is not None and not decision.approved:
                dispatch_results.append(
                    ToolResult(
                        id=call.get("id") or call.get("tool_call_id") or str(uuid.uuid4()),
                        name=call.get("name", ""),
                        error=f"ApprovalDenied: {decision.note or 'denied'}",
                        metadata={"tenant_id": tenant_id, "approval_decision": decision.note or "denied"},
                    )
                )
                continue
            dispatch_results.append(
                await self.dispatcher.dispatch(
                    tenant_id=tenant_id,
                    toolset=toolset or self.dispatcher.default_toolset or "default",
                    name=call.get("name", ""),
                    arguments=call_arguments,
                    tool_call_id=call.get("id") or call.get("tool_call_id") or str(uuid.uuid4()),
                )
            )

        tool_turn = ToolLoopResult(
            turn_id=str(uuid.uuid4()),
            messages=tuple(messages),
            tool_results=tuple(dispatch_results),
            finish_reason="tool_calls",
            tenant_id=tenant_id,
            metadata={"session_id": session_id, "toolset": toolset, "turn": len(loop_results) + 1},
        )
        loop_results.append(tool_turn)
        finish_reason = "tool_calls"
        await self._emit(
            AuditEvent(
                event="agent.tool_calls.dispatched",
                session_id=session_id,
                agent=self.agent,
                tool_call_id=dispatch_results[0].id if dispatch_results else None,
                data={"tools": [r.name for r in dispatch_results], "turn": len(loop_results)},
                tenant_id=tenant_id,
            ),
            tenant_id=tenant_id,
        )

        # Append tool results as tool messages so the next completion sees them.
        for res in dispatch_results:
            messages.append(
                ChatMessage(
                    role="tool",
                    content=str(res.output) if res.error is None else (res.error or ""),
                    name=res.name,
                    tool_call_id=res.id,
                    metadata={"tenant_id": tenant_id, "tool_result": True, "error": res.error},
                )
            )

    await self._emit(
        AuditEvent(event="agent.loop.end", session_id=session_id, agent=self.agent, tenant_id=tenant_id),
        tenant_id=tenant_id,
    )

    return AgentRuntimeResult(
        response=final_response,
        loop_results=tuple(loop_results),
        tenant_id=tenant_id,
        metadata={"session_id": session_id, "agent": self.agent},
    )

stream_tokens(*, request: ChatRequest, tenant_id: Optional[str] = None, toolset: Optional[str] = None) -> AsyncIterator[str] async

Stream incremental assistant tokens from the provider.

Calls provider.stream (real SSE streaming) and re-emits the partial content of each incremental :class:ChatResponse as it arrives, so callers see the answer grow token by token.

Source code in src/ciel/runtime/__init__.py 
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
async def stream_tokens(
    self,
    *,
    request: ChatRequest,
    tenant_id: Optional[str] = None,
    toolset: Optional[str] = None,
) -> AsyncIterator[str]:
    """Stream incremental assistant tokens from the provider.

    Calls ``provider.stream`` (real SSE streaming) and re-emits the
    partial ``content`` of each incremental :class:`ChatResponse` as it
    arrives, so callers see the answer grow token by token.
    """
    chunks = await self.provider.stream(request=request)
    prior = ""
    for chunk in chunks:
        content = chunk.choice.message.text()
        if content != prior:
            yield content
            prior = content

DefaultToolDispatcher

Dispatch tool requests to a configured ToolProvider.

Source code in src/ciel/runtime/__init__.py 
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
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
class DefaultToolDispatcher:
    """Dispatch tool requests to a configured ToolProvider."""

    provider: ToolProvider
    default_toolset: Optional[str]

    def __init__(self, provider: ToolProvider, default_toolset: Optional[str] = None) -> None:
        self.provider = provider
        self.default_toolset = default_toolset or getattr(provider.registry, "default_toolset", None)

    async def dispatch(
        self,
        *,
        tenant_id: Optional[str] = None,
        toolset: Optional[str] = None,
        name: str,
        arguments: Dict[str, Any],
        tool_call_id: str,
    ) -> ToolResult:
        result = await self.provider.execute(
            tenant_id=tenant_id,
            toolset=toolset or self.default_toolset or "default",
            name=name,
            arguments=arguments,
            tool_call_id=tool_call_id,
        )
        result.metadata.setdefault("tenant_id", tenant_id)
        return result

    async def dispatch_all(
        self,
        *,
        tenant_id: Optional[str] = None,
        toolset: Optional[str] = None,
        calls: Sequence[Dict[str, Any]],
    ):
        results: List[ToolResult] = []
        for call in calls:
            call_tenant_id = tenant_id or call.get("metadata", {}).get("tenant_id")
            results.append(
                await self.dispatch(
                    tenant_id=call_tenant_id,
                    toolset=toolset or self.default_toolset or "default",
                    name=call["name"],
                    arguments=call.get("arguments", {}),
                    tool_call_id=call.get("id") or call.get("tool_call_id") or str(uuid.uuid4()),
                )
            )
        return results

ToolProvider dataclass

Concrete tool provider used by the runtime's tool dispatcher.

Source code in src/ciel/runtime/__init__.py 
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
72
73
74
75
76
77
78
79
80
81
82
83
84
85
@dataclass(frozen=True)
class ToolProvider:
    """Concrete tool provider used by the runtime's tool dispatcher."""

    registry: ToolRegistry
    require_tenant_on_execution: bool = True

    async def tool_specs(self, tenant_id: Optional[str], toolset: str) -> Sequence[ToolSpec]:
        return tuple(getattr(self.registry, "_toolsets", {}).get(toolset, ToolsetSchema(name=toolset, description="")).tools)

    async def execute(
        self,
        *,
        tenant_id: Optional[str],
        toolset: Optional[str],
        name: str,
        arguments: Dict[str, Any],
        tool_call_id: str,
    ) -> ToolResult:
        target_toolset = toolset or (self.registry.default_toolset or "default")
        if self.require_tenant_on_execution and not tenant_id:
            raise TenantRequired(f"tenant_id is required to execute tool '{name}' in toolset '{target_toolset}'.")
        tool = self.registry.get_tool(toolset=target_toolset, name=name)
        if tool is None:
            return ToolResult(id=tool_call_id, name=name, error=f"unknown tool: {target_toolset}.{name}", metadata={"tenant_id": tenant_id})
        if getattr(tool, "required_tenant", False) and not tenant_id:
            raise TenantRequired(f"tool '{name}' requires tenant_id, but none was provided.")
        if tool.callable_ is None:
            output = {"arguments": arguments, "description": tool.spec.description}
            return ToolResult(id=tool_call_id, name=name, output=output, metadata={"tenant_id": tenant_id})
        # Official tool callable contract:
        #   callable_(arguments: dict, *, tool_call_id: str, tenant_id: str | None) -> ToolResult | dict | Any
        try:
            result = tool.callable_(arguments, tool_call_id=tool_call_id, tenant_id=tenant_id)
            if inspect.isawaitable(result):
                result = await result
        except Exception as exc:  # noqa: BLE001 — surface tool errors as ToolResult
            return ToolResult(id=tool_call_id, name=name, error=f"{type(exc).__name__}: {exc}", metadata={"tenant_id": tenant_id})
        if isinstance(result, ToolResult):
            if not result.metadata.get("tenant_id"):
                result.metadata["tenant_id"] = tenant_id
            return result
        return ToolResult(id=tool_call_id, name=name, output=result, metadata={"tenant_id": tenant_id})

ToolRegistry

Source code in src/ciel/runtime/tools.py 
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
class ToolRegistry:
    def __init__(self, *, default_toolset: Optional[str] = None) -> None:
        self._toolsets: Dict[str, ToolsetSchema] = {}
        self._tools: Dict[str, Dict[str, Tool]] = {}
        self._tenant_tools: Dict[str, Dict[str, Dict[str, Tool]]] = {}
        self.default_toolset = default_toolset

    def register_toolset(self, schema: ToolsetSchema) -> None:
        self._toolsets[schema.name] = schema
        self._tools.setdefault(schema.name, {})
        self._tenant_tools.setdefault(schema.name, {})
        for tool in schema.tools:
            self._tools[schema.name][tool.name] = Tool(spec=tool, required_tenant=schema.require_tenant)

    def register_tool(self, toolset, tool, *, tenant_id=None):
        if isinstance(toolset, ToolsetSchema):
            schema = toolset
            toolset = schema.name
            self._toolsets[schema.name] = schema
            self._tools.setdefault(schema.name, {})
            self._tenant_tools.setdefault(schema.name, {})
            for tool in schema.tools:
                self._tools[schema.name][tool.name] = Tool(spec=tool, required_tenant=schema.require_tenant)
            return
        schema = self._toolsets.get(toolset)
        require_tenant = schema.require_tenant if schema is not None else tool.required_tenant
        tool_obj = Tool(spec=tool.spec, callable_=tool.callable_, metadata=tool.metadata, required_tenant=tool.required_tenant or require_tenant)
        if schema is None:
            schema = ToolsetSchema(name=toolset, description="", require_tenant=require_tenant)
            self._toolsets[toolset] = schema
        self._tools.setdefault(toolset, {})
        self._tools[toolset][tool.spec.name] = tool_obj
        # Keep the schema's tool list in sync so get_toolset_schema/export_schema reflect registered tools.
        self._toolsets[toolset] = replace(schema, tools=tuple(t.spec for t in self._tools[toolset].values()))
        if tenant_id:
            self._tenant_tools.setdefault(toolset, {}).setdefault(tenant_id, {})[tool.spec.name] = tool_obj

    def get_toolset(self, name):
        return self._toolsets.get(name)

    def get_tool(self, toolset: str, name: str, tenant_id: Optional[str] = None):
        if tenant_id:
            tenant_tools = self._tenant_tools.get(toolset, {}).get(tenant_id)
            if tenant_tools is None:
                raise ValueError(f"tenant_id='{tenant_id}' has no mapped tools for toolset='{toolset}'")
            tool = tenant_tools.get(name)
            if tool is not None:
                return tool
        tools = self._tools.get(toolset)
        if not tools:
            return None
        return tools.get(name)

    def toolset_names(self) -> Sequence[str]:
        return tuple(self._toolsets.keys())

    def tool_names(self, toolset: str) -> Sequence[str]:
        return tuple(self._tools.get(toolset, {}).keys())

    def export_schema(self, toolset: str) -> Dict[str, Any]:
        schema = self._toolsets.get(toolset)
        if schema is None:
            raise KeyError(f"unknown toolset: {toolset}")
        return schema.to_json()

    async def lookup(self, *, tenant_id: Optional[str], toolset: str) -> Sequence[Tool]:
        schema = self._toolsets.get(toolset)
        if schema is None:
            return ()
        effective_tenant = schema.tenant_for(caller_tenant_id=tenant_id)
        if effective_tenant:
            tools = self._tenant_tools.get(toolset, {}).get(effective_tenant)
            if tools:
                return tuple(tools.values())
        return tuple(self._tools.get(toolset, {}).values())

WebhookAdapter

Bases: MessagingAdapter

Messaging adapter that ingests messages via an HTTP webhook endpoint.

Incoming webhook payloads are validated, converted to :class:Message instances, and enqueued on an internal :class:asyncio.Queue. The :meth:receive async generator drains the queue and yields messages to consumers.

Typical usage alongside an HTTP framework::

adapter = WebhookAdapter()

@app.post("/webhook")
async def webhook(request):
    payload = await request.json()
    return await adapter.handle_webhook(payload)

async for message in adapter.receive():
    ...
Source code in src/ciel/gateway/adapter.py 
 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
128
129
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
class WebhookAdapter(MessagingAdapter):
    """Messaging adapter that ingests messages via an HTTP webhook endpoint.

    Incoming webhook payloads are validated, converted to :class:`Message`
    instances, and enqueued on an internal :class:`asyncio.Queue`. The
    :meth:`receive` async generator drains the queue and yields messages
    to consumers.

    Typical usage alongside an HTTP framework::

        adapter = WebhookAdapter()

        @app.post("/webhook")
        async def webhook(request):
            payload = await request.json()
            return await adapter.handle_webhook(payload)

        async for message in adapter.receive():
            ...
    """

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

    # -- public API -------------------------------------------------

    async def handle_webhook(self, payload: dict) -> dict:
        """Validate a webhook payload, enqueue a :class:`Message`, and respond.

        Expected payload shape::

            {"id": "...", "channel": "...", "sender": "...",
             "content": "...", "metadata": {...}}

        ``channel`` and ``content`` are required. ``id`` is generated
        via :func:`uuid.uuid4` when absent. ``sender`` defaults to
        ``None`` and ``metadata`` defaults to an empty dict.

        Args:
            payload: Raw deserialized webhook body.

        Returns:
            ``{"status": "accepted", "id": <message id>}`` on success,
            or ``{"status": "rejected", "error": "missing required
            field: <field>"}`` when validation fails.
        """
        channel = payload.get("channel")
        content = payload.get("content")

        if not channel:
            return {"status": "rejected", "error": "missing required field: channel"}
        if not content:
            return {"status": "rejected", "error": "missing required field: content"}

        message = Message(
            id=payload.get("id") or str(uuid.uuid4()),
            channel=channel,
            sender=payload.get("sender"),
            content=content,
            metadata=payload.get("metadata") or {},
        )

        await self._queue.put(message)

        return {"status": "accepted", "id": message.id}

    async def receive(self) -> AsyncIterator[Message]:
        """Async generator yielding messages from the internal queue.

        Blocks indefinitely on each :meth:`asyncio.Queue.get` call until
        a message becomes available, then yields it. The loop runs
        forever — the caller is responsible for cancellation/timeout.
        """
        while True:
            message = await self._queue.get()
            yield message

    async def receive_internal(self) -> Message:
        """Drain and return a single message from the queue.

        Convenience method for consumers that prefer one-at-a-time
        polling over the async generator. Blocks until a message is
        available.

        Returns:
            The next :class:`Message` in the queue.
        """
        return await self._queue.get()

    async def send(self, message: Message) -> None:
        """Outbound send — not implemented for the webhook adapter.

        The webhook adapter is inbound-only by design. Subclass or
        compose with an outbound adapter if delivery is needed.
        """
        raise NotImplementedError

handle_webhook(payload: dict) -> dict async

Validate a webhook payload, enqueue a :class:Message, and respond.

Expected payload shape::

{"id": "...", "channel": "...", "sender": "...",
 "content": "...", "metadata": {...}}

channel and content are required. id is generated via :func:uuid.uuid4 when absent. sender defaults to None and metadata defaults to an empty dict.

Parameters:

Name Type Description Default
payload dict

Raw deserialized webhook body.

 required

Returns:

Name Type Description
dict

{"status": "accepted", "id": <message id>} on success,
dict

or ``{"status": "rejected", "error": "missing required
field dict

"}`` when validation fails.
Source code in src/ciel/gateway/adapter.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
async def handle_webhook(self, payload: dict) -> dict:
    """Validate a webhook payload, enqueue a :class:`Message`, and respond.

    Expected payload shape::

        {"id": "...", "channel": "...", "sender": "...",
         "content": "...", "metadata": {...}}

    ``channel`` and ``content`` are required. ``id`` is generated
    via :func:`uuid.uuid4` when absent. ``sender`` defaults to
    ``None`` and ``metadata`` defaults to an empty dict.

    Args:
        payload: Raw deserialized webhook body.

    Returns:
        ``{"status": "accepted", "id": <message id>}`` on success,
        or ``{"status": "rejected", "error": "missing required
        field: <field>"}`` when validation fails.
    """
    channel = payload.get("channel")
    content = payload.get("content")

    if not channel:
        return {"status": "rejected", "error": "missing required field: channel"}
    if not content:
        return {"status": "rejected", "error": "missing required field: content"}

    message = Message(
        id=payload.get("id") or str(uuid.uuid4()),
        channel=channel,
        sender=payload.get("sender"),
        content=content,
        metadata=payload.get("metadata") or {},
    )

    await self._queue.put(message)

    return {"status": "accepted", "id": message.id}

receive() -> AsyncIterator[Message] async

Async generator yielding messages from the internal queue.

Blocks indefinitely on each :meth:asyncio.Queue.get call until a message becomes available, then yields it. The loop runs forever — the caller is responsible for cancellation/timeout.

Source code in src/ciel/gateway/adapter.py 
150
151
152
153
154
155
156
157
158
159
async def receive(self) -> AsyncIterator[Message]:
    """Async generator yielding messages from the internal queue.

    Blocks indefinitely on each :meth:`asyncio.Queue.get` call until
    a message becomes available, then yields it. The loop runs
    forever — the caller is responsible for cancellation/timeout.
    """
    while True:
        message = await self._queue.get()
        yield message

receive_internal() -> Message async

Drain and return a single message from the queue.

Convenience method for consumers that prefer one-at-a-time polling over the async generator. Blocks until a message is available.

Returns:

Type Description
Message

The next :class:Message in the queue.
Source code in src/ciel/gateway/adapter.py 
161
162
163
164
165
166
167
168
169
170
171
async def receive_internal(self) -> Message:
    """Drain and return a single message from the queue.

    Convenience method for consumers that prefer one-at-a-time
    polling over the async generator. Blocks until a message is
    available.

    Returns:
        The next :class:`Message` in the queue.
    """
    return await self._queue.get()

send(message: Message) -> None async

Outbound send — not implemented for the webhook adapter.

The webhook adapter is inbound-only by design. Subclass or compose with an outbound adapter if delivery is needed.

Source code in src/ciel/gateway/adapter.py 
173
174
175
176
177
178
179
async def send(self, message: Message) -> None:
    """Outbound send — not implemented for the webhook adapter.

    The webhook adapter is inbound-only by design. Subclass or
    compose with an outbound adapter if delivery is needed.
    """
    raise NotImplementedError

_EchoProvider

Offline provider used when no remote LLM endpoint is configured.

It returns an echo:<prompt> completion so the gateway can boot and serve health/tool endpoints without network access. It is intentionally deterministic and never makes outbound calls.

Source code in src/ciel/gateway/server.py 
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 _EchoProvider:
    """Offline provider used when no remote LLM endpoint is configured.

    It returns an ``echo:<prompt>`` completion so the gateway can boot and
    serve health/tool endpoints without network access. It is intentionally
    deterministic and never makes outbound calls.
    """

    provider_name = "echo"

    async def complete(self, request) -> ChatResponse:
        prompt = request.messages[-1].text() if request.messages else ""
        return ChatResponse(
            choice=ChatChoice(
                message=ChatMessage(role="assistant", content=f"echo:{prompt}"),
                finish_reason="stop",
            ),
            metadata={},
        )

    async def stream(self, request):  # pragma: no cover - parity with runtime contract
        return (await self.complete(request),)

    async def models(self):  # pragma: no cover - not exercised by gateway
        from ciel.providers import ModelInfo

        return [ModelInfo(id="echo", provider=self.provider_name)]

_build_chat_provider() -> object

Source code in src/ciel/gateway/server.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
def _build_chat_provider() -> object:
    provider_url = os.getenv("CIEL_PROVIDER_URL")
    if provider_url:
        from ciel.providers import OpenAICompatibleProvider

        logger.info("ciel serve: using remote provider %s", provider_url)
        return OpenAICompatibleProvider(
            base_url=provider_url,
            api_key=os.getenv("CIEL_API_KEY"),
            default_model=os.getenv("CIEL_MODEL"),
        )
    logger.info("ciel serve: no CIEL_PROVIDER_URL set; booting with offline echo provider")
    return _EchoProvider()

create_control_app(*, runtime: DefaultAgentRuntime, registry: Optional[ProviderRegistry] = None, audit_sink: Any = None, tenant_id: Optional[str] = None, api_key: Optional[str] = None, board_db_path: Optional[str] = None) -> FastAPI

Build a FastAPI control plane for the Ciel agent runtime.

Parameters

runtime: A :class:ciel.runtime.DefaultAgentRuntime used to run agent loops and dispatch tools. Tool dispatch goes through runtime.dispatcher. El endpoint POST /v1/agent/run/stream usa runtime.stream_tokens para emitir Server-Sent Events (SSE) con los fragmentos incrementales del assistant y cierra con data: [DONE]. registry: Optional :class:ciel.providers.ProviderRegistry surfaced via the GET /info endpoint. If omitted, an empty provider list is reported. audit_sink: Optional audit sink retained for observability hooks. Not directly invoked here — the runtime already emits audit events. tenant_id: Default tenant propagated to the runtime when a request does not provide its own tenant_id. api_key: Optional transport API key. When None (the default), the CIEL_API_KEY environment variable is consulted. If a key is configured (explicitly or via env), protected routes require a valid key via Authorization: Bearer *** orX-API-Key; otherwise the dependency is a no-op (open mode). board_db_path: Optional path to a SQLite database for the kanban board. When set (o bien vía la variable de entornoCIEL_BOARD_DB), el board se monta sobre SQLite para que/v1/board/list`` vea las tareas creadas por el CLI y viceversa. Si no se configura ninguna ruta, el board queda en memoria (comportamiento legacy, útil para smoke tests/offline).

Source code in src/ciel/gateway/base.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
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
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
def create_control_app(
    *,
    runtime: DefaultAgentRuntime,
    registry: Optional[ProviderRegistry] = None,
    audit_sink: Any = None,
    tenant_id: Optional[str] = None,
    api_key: Optional[str] = None,
    board_db_path: Optional[str] = None,
) -> FastAPI:
    """Build a FastAPI control plane for the Ciel agent runtime.

    Parameters
    ----------
    runtime:
        A :class:`ciel.runtime.DefaultAgentRuntime` used to run agent loops and
        dispatch tools. Tool dispatch goes through ``runtime.dispatcher``.
        El endpoint ``POST /v1/agent/run/stream`` usa
        ``runtime.stream_tokens`` para emitir Server-Sent Events (SSE) con los
        fragmentos incrementales del assistant y cierra con ``data: [DONE]``.
    registry:
        Optional :class:`ciel.providers.ProviderRegistry` surfaced via the
        ``GET /info`` endpoint. If omitted, an empty provider list is reported.
    audit_sink:
        Optional audit sink retained for observability hooks. Not directly
        invoked here — the runtime already emits audit events.
    tenant_id:
        Default tenant propagated to the runtime when a request does not
        provide its own ``tenant_id``.
    api_key:
        Optional transport API key. When ``None`` (the default), the
        ``CIEL_API_KEY`` environment variable is consulted. If a key is
        configured (explicitly or via env), protected routes require a valid
        key via ``Authorization: Bearer *** or ``X-API-Key``; otherwise
        the dependency is a no-op (open mode).
    board_db_path:
        Optional path to a SQLite database for the kanban board. When set (o
        bien vía la variable de entorno ``CIEL_BOARD_DB``), el board se monta
        sobre SQLite para que ``/v1/board/list`` vea las tareas creadas por el
        CLI y viceversa. Si no se configura ninguna ruta, el board queda en
        memoria (comportamiento legacy, útil para smoke tests/offline).
    """
    from ciel.gateway.auth import Depends, make_auth_dependency

    api_key_guard = Depends(make_auth_dependency(expected_key=api_key))

    app = FastAPI(
        title="Ciel Control Gateway",
        version=__version__,
        description="HTTP control plane for the Ciel agent runtime.",
    )
    # Attach references for tests / introspection.
    app.state.runtime = runtime
    app.state.registry = registry
    app.state.audit_sink = audit_sink
    app.state.default_tenant_id = tenant_id

    # Resolve a live board instance if orchestration is available.
    # Si se pasa board_db_path o existe CIEL_BOARD_DB, el board se monta sobre
    # SQLite para compartir estado con el CLI; en caso contrario queda en
    # memoria (fallback para smoke tests/offline).
    board: Optional[KanbanBoard] = None
    resolved_board_db = board_db_path or os.environ.get("CIEL_BOARD_DB")
    try:
        board = KanbanBoard(path=resolved_board_db)  # None => en memoria.
    except Exception:  # pragma: no cover - defensive guard
        board = None
    app.state.board = board

    def _resolve_tenant(request_tenant: Optional[str]) -> Optional[str]:
        return request_tenant or tenant_id

    @app.get("/health", response_model=HealthResponse, tags=["meta"])
    async def health() -> HealthResponse:
        return HealthResponse(version=__version__)

    # --- Fase 14 / F16: health reales (liveness vs readiness) ---------------
    @app.get("/healthz", response_model=LivenessResponse, tags=["meta"])
    async def healthz() -> LivenessResponse:
        # Liveness: el proceso está vivo. No depende de backends remotos.
        return LivenessResponse(status="alive", version=__version__)

    @app.get("/readyz", response_model=ReadinessResponse, tags=["meta"])
    async def readyz(request: Request) -> ReadinessResponse:
        # Readiness: el gateway puede atender tráfico. Depende de que el
        # StateBackend compartido esté conectado y migrado.
        backend = getattr(request.app.state, "state_backend", None)
        if backend is None:
            # Sin backend cableado => no listo para multi-réplica.
            return ReadinessResponse(
                status="not_ready",
                version=__version__,
                backend="none",
                backend_ready=False,
                details={"reason": "state_backend not wired"},
            )
        ready = backend.is_ready()
        return ReadinessResponse(
            status="ready" if ready else "not_ready",
            version=__version__,
            backend=backend.backend_type,
            backend_ready=ready,
            details={"ready": ready},
        )

    @app.get("/info", response_model=InfoResponse, tags=["meta"])
    async def info() -> InfoResponse:
        providers: List[ProviderInfo] = []
        if registry is not None:
            for name in registry.available():
                provider_name: Optional[str] = None
                try:
                    provider = registry.get(name)
                    provider_name = getattr(provider, "provider_name", None)
                except Exception:
                    provider_name = None
                providers.append(ProviderInfo(name=name, provider_name=provider_name))
        return InfoResponse(
            version=__version__,
            providers=providers,
            default_tenant=tenant_id,
        )

    @app.post(
        "/v1/agent/run",
        response_model=AgentRunResponse,
        tags=["agent"],
        dependencies=[api_key_guard],
    )
    async def agent_run(body: AgentRunRequest) -> AgentRunResponse:
        effective_tenant = _resolve_tenant(body.tenant_id)
        if effective_tenant is None:
            raise HTTPException(
                status_code=400,
                detail="tenant_id is required (multi-tenancy is enforced). "
                "Provide it in the request or configure a default tenant in the gateway.",
            )
        session_id = body.session_id or str(uuid.uuid4())
        request = _build_chat_request(body, session_id, effective_tenant)
        try:
            result = await runtime.run_agent_loop(
                request=request,
                tenant_id=effective_tenant,
                toolset=body.toolset,
            )
        except Exception as exc:
            logger.exception("agent.run failed: %s", exc)
            raise HTTPException(status_code=500, detail=f"agent loop failed: {exc}") from exc
        return _serialize_chat_response(result)

    def _build_chat_request(body: AgentRunRequest, session_id: str, effective_tenant: Optional[str]) -> ChatRequest:
        extra: Dict[str, Any] = dict(body.extra or {})
        extra.setdefault("session_id", session_id)
        if effective_tenant is not None:
            extra.setdefault("tenant_id", effective_tenant)
        return ChatRequest(
            messages=(ChatMessage(role="user", content=body.prompt),),
            tools=(),
            model=body.model,
            temperature=body.temperature,
            max_tokens=body.max_tokens,
            extra=extra,
        )

    @app.post(
        "/v1/agent/run/stream",
        tags=["agent"],
        dependencies=[api_key_guard],
    )
    async def agent_run_stream(body: AgentRunRequest) -> StreamingResponse:
        """Stream the assistant completion as Server-Sent Events.

        Emite cada fragmento incremental del assistant como un evento SSE
        ``data: <token>`` y cierra la respuesta con ``data: [DONE]``.
        Reutiliza ``runtime.stream_tokens`` (que a su vez llama a
        ``provider.stream`` para hacer SSE real cuando el proveedor lo soporta).

        El ``tenant_id`` es obligatorio, igual que en ``/v1/agent/run``.
        """
        effective_tenant = _resolve_tenant(body.tenant_id)
        if effective_tenant is None:
            raise HTTPException(
                status_code=400,
                detail="tenant_id is required (multi-tenancy is enforced). "
                "Provide it in the request or configure a default tenant in the gateway.",
            )
        session_id = body.session_id or str(uuid.uuid4())

        async def _event_generator():
            try:
                request = _build_chat_request(body, session_id, effective_tenant)
                async for token in runtime.stream_tokens(
                    request=request,
                    tenant_id=effective_tenant,
                    toolset=body.toolset,
                ):
                    yield f"data: {token}\n\n"
            except Exception as exc:  # pragma: no cover - defensive guard
                logger.exception("agent.run.stream failed: %s", exc)
                yield f"data: [ERROR] {exc}\n\n"
            yield "data: [DONE]\n\n"

        return StreamingResponse(_event_generator(), media_type="text/event-stream")

    @app.post(
        "/v1/tools/{toolset}/{name}",
        response_model=ToolInvokeResponse,
        tags=["tools"],
        dependencies=[api_key_guard],
    )
    async def tool_invoke(toolset: str, name: str, body: ToolInvokeRequest) -> ToolInvokeResponse:
        dispatcher: Optional[DefaultToolDispatcher] = getattr(runtime, "dispatcher", None)
        if dispatcher is None:
            raise HTTPException(status_code=500, detail="runtime has no tool dispatcher")
        effective_tenant = _resolve_tenant(body.tenant_id)
        if effective_tenant is None:
            raise HTTPException(
                status_code=400,
                detail="tenant_id is required to dispatch tools (multi-tenancy is enforced).",
            )
        tool_call_id = body.tool_call_id or str(uuid.uuid4())
        try:
            result = await dispatcher.dispatch(
                tenant_id=effective_tenant,
                toolset=toolset,
                name=name,
                arguments=dict(body.arguments or {}),
                tool_call_id=tool_call_id,
            )
        except Exception as exc:
            logger.exception("tool.invoke failed: %s", exc)
            raise HTTPException(status_code=500, detail=f"tool dispatch failed: {exc}") from exc
        return ToolInvokeResponse(
            id=getattr(result, "id", None),
            name=getattr(result, "name", name),
            output=getattr(result, "output", None),
            error=getattr(result, "error", None),
            metadata=dict(getattr(result, "metadata", {}) or {}),
        )

    @app.get(
        "/v1/board/list",
        response_model=BoardListResponse,
        tags=["board"],
        dependencies=[api_key_guard],
    )
    async def board_list(tenant_id: Optional[str] = None) -> BoardListResponse:
        effective_tenant = _resolve_tenant(tenant_id)
        if board is None:
            return BoardListResponse(tasks=[])
        return BoardListResponse(tasks=_list_board_tasks(board, effective_tenant))

    return app

make_app(*, tenant_id: Optional[str] = None, include_mcp: bool = True, include_webhook: bool = True, state_backend: Optional[object] = None) -> FastAPI

Compose the full Ciel gateway into a single FastAPI app.

Parameters

tenant_id: Default tenant propagated to the runtime when a request omits its own tenant_id. Falls back to the CIEL_TENANT environment variable. include_mcp: Mount the MCP host app at /mcp. include_webhook: Mount the webhook messaging router at /v1/messaging/webhook. state_backend: Optional pre-built :class:StateBackend (Fase 14 / F15). When omitted, one is built from CIEL_STATE_BACKEND / CIEL_STATE_DSN (default SQLite). Injecting a shared backend lets tests and multi-replica deployments reuse the same state source.

Source code in src/ciel/gateway/server.py 
 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
128
129
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
181
182
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
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
def make_app(
    *,
    tenant_id: Optional[str] = None,
    include_mcp: bool = True,
    include_webhook: bool = True,
    state_backend: Optional[object] = None,
) -> FastAPI:
    """Compose the full Ciel gateway into a single FastAPI app.

    Parameters
    ----------
    tenant_id:
        Default tenant propagated to the runtime when a request omits its own
        ``tenant_id``. Falls back to the ``CIEL_TENANT`` environment variable.
    include_mcp:
        Mount the MCP host app at ``/mcp``.
    include_webhook:
        Mount the webhook messaging router at ``/v1/messaging/webhook``.
    state_backend:
        Optional pre-built :class:`StateBackend` (Fase 14 / F15). When omitted,
        one is built from ``CIEL_STATE_BACKEND`` / ``CIEL_STATE_DSN`` (default
        SQLite). Injecting a shared backend lets tests and multi-replica
        deployments reuse the same state source.
    """
    tenant_id = tenant_id or os.getenv("CIEL_TENANT")

    # --- runtime wiring ----------------------------------------------------
    chat_provider = _build_chat_provider()
    registry = ToolRegistry(default_toolset="default")
    tool_provider = ToolProvider(registry=registry, require_tenant_on_execution=True)
    dispatcher = DefaultToolDispatcher(provider=tool_provider, default_toolset="default")
    runtime = DefaultAgentRuntime(provider=chat_provider, dispatcher=dispatcher)

    # --- Fase 14 / F15: StateBackend compartido (multi-réplica) ------------
    # Construye el backend de estado según CIEL_STATE_BACKEND / CIEL_STATE_DSN
    # (default: SQLite local offline-safe). Los stores de checkpoint/session
    # (F16) lo consumen para compartir estado entre réplicas. Expuesto en
    # app.state para los health endpoints (/readyz) y resume-on-startup.
    from ciel.runtime.state_backend import StateBackend, build_state_backend

    if state_backend is None:
        state_backend = build_state_backend()
    else:
        # Validación defensiva: debe cumplir la interfaz StateBackend.
        assert isinstance(state_backend, StateBackend), "state_backend debe ser un StateBackend"


    # Lazy import to avoid a circular import with the ``ciel.gateway`` package
    # (server.py is imported by gateway/__init__.py; the two helpers below are
    # defined there and only needed at call time, after the package is loaded).
    from ciel.gateway import create_webhook_router, mount_mcp_app

    # --- control plane (root app) -----------------------------------------
    app = create_control_app(runtime=runtime, registry=None, tenant_id=tenant_id)
    app.title = "Ciel Agent Framework Gateway"
    app.version = __version__

    # --- MCP host (mounted sub-application) -------------------------------
    if include_mcp:
        # Expose the JSON-RPC endpoint at the mount root so the resulting route
        # is ``POST /mcp/`` and health at ``GET /mcp/health``.
        mcp_app = mount_mcp_app(dispatcher=dispatcher, tenant_id=tenant_id, path="/")
        app.mount("/mcp", mcp_app)
        app.state.mcp_mounted = True
    else:
        app.state.mcp_mounted = False

    # --- webhook messaging adapter ----------------------------------------
    if include_webhook:
        adapter = WebhookAdapter()
        app.include_router(create_webhook_router(adapter))
        app.state.webhook_adapter = adapter

    # --- Fase 8: adapters de canal Teams/Discord/Web UI (offline-safe) -----
    # Se montan solo si no estamos en modo ultra-ligero (siempre por defecto).
    # Cada router enqueuea eventos entrantes en su adapter; el runtime los
    # consume vía ``adapter.receive()``. No requieren red para arrancar.
    try:
        from ciel.adapters import DiscordAdapter, TeamsAdapter, WebUIAdapter
        from ciel.gateway.messaging import (
            create_discord_webhook_router,
            create_teams_webhook_router,
            create_webui_router,
        )

        teams_adapter = TeamsAdapter(webhook_url=os.getenv("CIEL_TEAMS_WEBHOOK"))
        discord_adapter = DiscordAdapter(webhook_url=os.getenv("CIEL_DISCORD_WEBHOOK"))
        webui_adapter = WebUIAdapter()
        app.include_router(create_teams_webhook_router(teams_adapter))
        app.include_router(create_discord_webhook_router(discord_adapter))
        app.include_router(create_webui_router(webui_adapter))
        app.state.teams_adapter = teams_adapter
        app.state.discord_adapter = discord_adapter
        app.state.webui_adapter = webui_adapter
        logger.info("ciel serve: mounted Teams/Discord/WebUI messaging routers")
    except Exception as exc:  # pragma: no cover - defensive
        logger.warning("ciel serve: channel adapters not mounted: %s", exc)

    # --- Fase 13 / F19: Ciel Studio dashboard (offline-safe) --------
    # Expone /v1/studio con el snapshot de sesiones/loops. Usa el store
    # singleton de studio (compartido con install_studio_support).
    try:
        from ciel.studio import create_studio_router, get_studio_store

        studio_store = get_studio_store()
        app.include_router(create_studio_router(store=studio_store))
        app.state.studio_store = studio_store
        logger.info("ciel serve: mounted Ciel Studio dashboard at /v1/studio")
    except Exception as exc:  # pragma: no cover - defensive
        logger.warning("ciel serve: Ciel Studio not mounted: %s", exc)

    # --- Fase 13 / F20: Ciel Studio graph trace + replay ------------
    try:
        from ciel.studio_trace import create_trace_router, get_trace_store

        trace_store = get_trace_store()
        app.include_router(create_trace_router(store=trace_store))
        app.state.trace_store = trace_store
        logger.info("ciel serve: mounted Ciel Studio trace at /v1/studio/trace")
    except Exception as exc:  # pragma: no cover - defensive
        logger.warning("ciel serve: Ciel Studio trace not mounted: %s", exc)

    # --- Fase 13 / F21: Ciel Studio cost dashboard -------------------
    try:
        from ciel.studio_cost import create_cost_router, get_cost_store

        cost_store = get_cost_store()
        app.include_router(create_cost_router(store=cost_store))
        app.state.cost_store = cost_store
        logger.info("ciel serve: mounted Ciel Studio cost at /v1/studio/cost")
    except Exception as exc:  # pragma: no cover - defensive
        logger.warning("ciel serve: Ciel Studio cost not mounted: %s", exc)

    # --- Prometheus /metrics endpoint (lenient) --------------------------
    # Only mounted when prometheus-client is available; otherwise the import
    # is skipped so the gateway still boots offline.
    try:
        from ciel.observability.metrics import PROM_AVAILABLE, metrics_handler

        if PROM_AVAILABLE:
            app.add_api_route(
                "/metrics", metrics_handler, methods=["GET"], include_in_schema=False
            )
            app.state.metrics_mounted = True
            logger.info("ciel serve: mounted /metrics (Prometheus)")
        else:  # pragma: no cover - depends on optional extra
            app.state.metrics_mounted = False
            logger.info("ciel serve: /metrics not mounted (prometheus-client absent)")
    except Exception as exc:  # pragma: no cover - defensive
        app.state.metrics_mounted = False
        logger.warning("ciel serve: /metrics mount skipped: %s", exc)

    # --- Fase 14 / F15: stores de resume sobre el StateBackend compartido -
    # Instancia los stores de checkpoint/session contra el backend para que
    # estén disponibles para resume-on-startup (F16) y otros componentes, sin
    # alterar el control plane existente. Multi-réplica: comparten state vía
    # el backend (Postgres en prod, SQLite local en dev).
    try:
        from ciel.orchestration.agent import EventLoopCheckpointStore
        from ciel.orchestration.graph import GraphCheckpointStore
        from ciel.orchestration.session import SessionStore
        from ciel.runtime.checkpoints import CheckpointStore

        checkpoint_store = CheckpointStore(state_backend)
        session_store = SessionStore(state_backend)
        graph_checkpoint_store = GraphCheckpointStore(state_backend)
        eventloop_checkpoint_store = EventLoopCheckpointStore(state_backend)
        app.state.state_backend = state_backend
        app.state.checkpoint_store = checkpoint_store
        app.state.session_store = session_store
        app.state.graph_checkpoint_store = graph_checkpoint_store
        app.state.eventloop_checkpoint_store = eventloop_checkpoint_store
        logger.info(
            "ciel serve: state backend '%s' listo (stores de resume instanciados)",
            state_backend.backend_type,
        )
    except Exception as exc:  # pragma: no cover - defensive
        app.state.state_backend = state_backend
        logger.warning("ciel serve: state stores not wired: %s", exc)

    return app

Generic messaging adapter for webhook-based HTTP message intake.

This module is pure Python + asyncio. It does not depend on FastAPI, ciel.runtime, or ciel.providers — it is runtime-agnostic so any HTTP server (FastAPI, Starlette, aiohttp, etc.) can plug into it.

__all__ = ['Message', 'MessagingAdapter', 'WebhookAdapter'] module-attribute

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

WebhookAdapter

Bases: MessagingAdapter

Messaging adapter that ingests messages via an HTTP webhook endpoint.

Incoming webhook payloads are validated, converted to :class:Message instances, and enqueued on an internal :class:asyncio.Queue. The :meth:receive async generator drains the queue and yields messages to consumers.

Typical usage alongside an HTTP framework::

adapter = WebhookAdapter()

@app.post("/webhook")
async def webhook(request):
    payload = await request.json()
    return await adapter.handle_webhook(payload)

async for message in adapter.receive():
    ...
Source code in src/ciel/gateway/adapter.py 
 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
128
129
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
class WebhookAdapter(MessagingAdapter):
    """Messaging adapter that ingests messages via an HTTP webhook endpoint.

    Incoming webhook payloads are validated, converted to :class:`Message`
    instances, and enqueued on an internal :class:`asyncio.Queue`. The
    :meth:`receive` async generator drains the queue and yields messages
    to consumers.

    Typical usage alongside an HTTP framework::

        adapter = WebhookAdapter()

        @app.post("/webhook")
        async def webhook(request):
            payload = await request.json()
            return await adapter.handle_webhook(payload)

        async for message in adapter.receive():
            ...
    """

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

    # -- public API -------------------------------------------------

    async def handle_webhook(self, payload: dict) -> dict:
        """Validate a webhook payload, enqueue a :class:`Message`, and respond.

        Expected payload shape::

            {"id": "...", "channel": "...", "sender": "...",
             "content": "...", "metadata": {...}}

        ``channel`` and ``content`` are required. ``id`` is generated
        via :func:`uuid.uuid4` when absent. ``sender`` defaults to
        ``None`` and ``metadata`` defaults to an empty dict.

        Args:
            payload: Raw deserialized webhook body.

        Returns:
            ``{"status": "accepted", "id": <message id>}`` on success,
            or ``{"status": "rejected", "error": "missing required
            field: <field>"}`` when validation fails.
        """
        channel = payload.get("channel")
        content = payload.get("content")

        if not channel:
            return {"status": "rejected", "error": "missing required field: channel"}
        if not content:
            return {"status": "rejected", "error": "missing required field: content"}

        message = Message(
            id=payload.get("id") or str(uuid.uuid4()),
            channel=channel,
            sender=payload.get("sender"),
            content=content,
            metadata=payload.get("metadata") or {},
        )

        await self._queue.put(message)

        return {"status": "accepted", "id": message.id}

    async def receive(self) -> AsyncIterator[Message]:
        """Async generator yielding messages from the internal queue.

        Blocks indefinitely on each :meth:`asyncio.Queue.get` call until
        a message becomes available, then yields it. The loop runs
        forever — the caller is responsible for cancellation/timeout.
        """
        while True:
            message = await self._queue.get()
            yield message

    async def receive_internal(self) -> Message:
        """Drain and return a single message from the queue.

        Convenience method for consumers that prefer one-at-a-time
        polling over the async generator. Blocks until a message is
        available.

        Returns:
            The next :class:`Message` in the queue.
        """
        return await self._queue.get()

    async def send(self, message: Message) -> None:
        """Outbound send — not implemented for the webhook adapter.

        The webhook adapter is inbound-only by design. Subclass or
        compose with an outbound adapter if delivery is needed.
        """
        raise NotImplementedError

handle_webhook(payload: dict) -> dict async

Validate a webhook payload, enqueue a :class:Message, and respond.

Expected payload shape::

{"id": "...", "channel": "...", "sender": "...",
 "content": "...", "metadata": {...}}

channel and content are required. id is generated via :func:uuid.uuid4 when absent. sender defaults to None and metadata defaults to an empty dict.

Parameters:

Name Type Description Default
payload dict

Raw deserialized webhook body.

 required

Returns:

Name Type Description
dict

{"status": "accepted", "id": <message id>} on success,
dict

or ``{"status": "rejected", "error": "missing required
field dict

"}`` when validation fails.
Source code in src/ciel/gateway/adapter.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
async def handle_webhook(self, payload: dict) -> dict:
    """Validate a webhook payload, enqueue a :class:`Message`, and respond.

    Expected payload shape::

        {"id": "...", "channel": "...", "sender": "...",
         "content": "...", "metadata": {...}}

    ``channel`` and ``content`` are required. ``id`` is generated
    via :func:`uuid.uuid4` when absent. ``sender`` defaults to
    ``None`` and ``metadata`` defaults to an empty dict.

    Args:
        payload: Raw deserialized webhook body.

    Returns:
        ``{"status": "accepted", "id": <message id>}`` on success,
        or ``{"status": "rejected", "error": "missing required
        field: <field>"}`` when validation fails.
    """
    channel = payload.get("channel")
    content = payload.get("content")

    if not channel:
        return {"status": "rejected", "error": "missing required field: channel"}
    if not content:
        return {"status": "rejected", "error": "missing required field: content"}

    message = Message(
        id=payload.get("id") or str(uuid.uuid4()),
        channel=channel,
        sender=payload.get("sender"),
        content=content,
        metadata=payload.get("metadata") or {},
    )

    await self._queue.put(message)

    return {"status": "accepted", "id": message.id}

receive() -> AsyncIterator[Message] async

Async generator yielding messages from the internal queue.

Blocks indefinitely on each :meth:asyncio.Queue.get call until a message becomes available, then yields it. The loop runs forever — the caller is responsible for cancellation/timeout.

Source code in src/ciel/gateway/adapter.py 
150
151
152
153
154
155
156
157
158
159
async def receive(self) -> AsyncIterator[Message]:
    """Async generator yielding messages from the internal queue.

    Blocks indefinitely on each :meth:`asyncio.Queue.get` call until
    a message becomes available, then yields it. The loop runs
    forever — the caller is responsible for cancellation/timeout.
    """
    while True:
        message = await self._queue.get()
        yield message

receive_internal() -> Message async

Drain and return a single message from the queue.

Convenience method for consumers that prefer one-at-a-time polling over the async generator. Blocks until a message is available.

Returns:

Type Description
Message

The next :class:Message in the queue.
Source code in src/ciel/gateway/adapter.py 
161
162
163
164
165
166
167
168
169
170
171
async def receive_internal(self) -> Message:
    """Drain and return a single message from the queue.

    Convenience method for consumers that prefer one-at-a-time
    polling over the async generator. Blocks until a message is
    available.

    Returns:
        The next :class:`Message` in the queue.
    """
    return await self._queue.get()

send(message: Message) -> None async

Outbound send — not implemented for the webhook adapter.

The webhook adapter is inbound-only by design. Subclass or compose with an outbound adapter if delivery is needed.

Source code in src/ciel/gateway/adapter.py 
173
174
175
176
177
178
179
async def send(self, message: Message) -> None:
    """Outbound send — not implemented for the webhook adapter.

    The webhook adapter is inbound-only by design. Subclass or
    compose with an outbound adapter if delivery is needed.
    """
    raise NotImplementedError

Bidirectional Slack messaging adapter.

This module provides :class:SlackAdapter, a concrete :class:~ciel.gateway.adapter.MessagingAdapter that can send messages to Slack over the real Web API (chat.postMessage) and receive inbound events via a simple polling loop.

The slack_sdk dependency is lenient: importing this module never crashes the package when the SDK is absent. Only the act of constructing a :class:SlackAdapter lazily imports slack_sdk (it is pulled in by the messaging extra). If send is invoked without the SDK installed, a clear :class:RuntimeError explaining how to install it is raised — but the rest of the package keeps working.

Requires Python >= 3.10 (uses X | None syntax).

_SLACK_SDK_AVAILABLE = True module-attribute

__all__ = ['SlackAdapter'] module-attribute

logger = logging.getLogger(__name__) module-attribute

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

SlackAdapter

Bases: MessagingAdapter

Bidirectional Slack adapter built on slack_sdk.WebClient.

Parameters:

Name Type Description Default
token str

A Slack bot/user token (xoxb-...). Stored and used to construct the :class:slack_sdk.WebClient.

 required
channel Optional[str]

Optional default channel (e.g. "#general" or a channel ID). Used by :meth:send when a :class:Message does not carry an explicit channel in its metadata.

 None
bot_user_id Optional[str]

Optional bot user id (U...). Reserved for inbound event filtering; not required for sending.

 None

The Slack WebClient is created lazily so that merely importing this module (or instantiating the adapter) is safe even when slack_sdk is not installed. The client is built on first use of :meth:send / :meth:receive.

Source code in src/ciel/gateway/adapter_slack.py 
 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
 72
 73
 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
128
129
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
181
182
183
184
185
186
187
class SlackAdapter(MessagingAdapter):
    """Bidirectional Slack adapter built on ``slack_sdk.WebClient``.

    Args:
        token: A Slack bot/user token (``xoxb-...``). Stored and used to
            construct the :class:`slack_sdk.WebClient`.
        channel: Optional default channel (e.g. ``"#general"`` or a channel
            ID). Used by :meth:`send` when a :class:`Message` does not carry
            an explicit channel in its ``metadata``.
        bot_user_id: Optional bot user id (``U...``). Reserved for inbound
            event filtering; not required for sending.

    The Slack WebClient is created lazily so that merely importing this module
    (or instantiating the adapter) is safe even when ``slack_sdk`` is not
    installed. The client is built on first use of :meth:`send` / :meth:`receive`.
    """

    def __init__(
        self,
        token: str,
        *,
        channel: Optional[str] = None,
        bot_user_id: Optional[str] = None,
    ) -> None:
        self.token = token
        self.channel = channel
        self.bot_user_id = bot_user_id
        self._client = None  # type: ignore[var-annotated]
        self._inbound: "asyncio.Queue[Message]" = asyncio.Queue()

    def enqueue(self, message: Message) -> None:
        """Push an inbound :class:`Message` onto the internal queue.

        Used by the Slack webhook router to feed real ``message.changed``
        events into the adapter; consumers can drain them via
        :meth:`receive` (which is overridden below to prefer the queue).
        """
        self._inbound.put_nowait(message)

    # -- internal helpers -------------------------------------------------

    def _get_client(self):
        """Return the lazily-constructed ``slack_sdk.WebClient``.

        Raises:
            RuntimeError: If ``slack_sdk`` is not installed.
        """
        if self._client is None:
            if not _SLACK_SDK_AVAILABLE:  # pragma: no cover - SDK missing path
                raise RuntimeError(
                    "slack_sdk is not installed. Install it with "
                    "`uv pip install slack-sdk` (extra: messaging) to use SlackAdapter."
                )
            from slack_sdk import WebClient

            self._client = WebClient(token=self.token)
        return self._client

    def _resolve_channel(self, message: Message) -> str:
        """Pick the destination channel for a message.

        Resolution order: ``message.channel`` (the canonical DTO field) ->
        ``message.metadata['channel']`` (convenience override) -> ``self.channel``.
        """
        channel = message.channel or message.metadata.get("channel") or self.channel
        if not channel:
            raise ValueError(
                "No channel resolved for Slack send: provide message.channel "
                "or pass channel= to SlackAdapter."
            )
        return channel

    # -- public API -------------------------------------------------------

    async def send(self, message: Message) -> None:
        """Send ``message.content`` to Slack via ``chat.postMessage``.

        The destination channel is resolved from ``message.metadata['channel']``
        falling back to the adapter's default ``channel``. A clear
        :class:`ValueError` is raised when no channel can be resolved.

        Network access is required for this call to succeed against the real
        Slack API; tests pass a mocked ``WebClient`` to verify behaviour
        without touching the network.

        Args:
            message: The :class:`Message` to deliver.
        """
        client = self._get_client()
        channel = self._resolve_channel(message)
        # chat_postMessage is synchronous in slack_sdk; run it in a thread so
        # this stays friendly to the asyncio event loop.
        loop = asyncio.get_running_loop()
        await loop.run_in_executor(
            None,
            lambda: client.chat_postMessage(channel=channel, text=message.text()),
        )
        logger.debug("SlackAdapter.send -> channel=%s len=%d", channel, len(message.text()))

    async def receive(self) -> AsyncIterator[Message]:
        """Yield inbound Slack messages via a simple polling loop.

        This is a lenient, best-effort implementation: it relies on Slack's
        ``conversations.history`` to pull recent messages and yields any that
        are not authored by the bot user. It polls every few seconds.

        The primary, fully-tested contract of this adapter is :meth:`send`.
        The receive loop is provided so the adapter is genuinely bidirectional,
        but it is intentionally simple and skips advanced cursor pagination and
        deduplication. Inbound event handling is better served by mounting the
        Slack webhook router (see :func:`create_slack_webhook_router`) onto the
        FastAPI app, which enqueues real ``message.changed`` events.

        The loop runs forever; the caller is responsible for cancellation.
        """
        client = self._get_client()
        if not self.channel:
            logger.warning(
                "SlackAdapter.receive: no default channel set; polling skipped."
            )
            return
        seen: set[str] = set()
        loop = asyncio.get_running_loop()
        while True:
            try:
                response = await loop.run_in_executor(
                    None,
                    lambda: client.conversations_history(channel=self.channel, limit=20),
                )
                for msg in response.get("messages", []):
                    msg_id = msg.get("ts")
                    user = msg.get("user")
                    if msg_id in seen:
                        continue
                    seen.add(msg_id)
                    if self.bot_user_id and user == self.bot_user_id:
                        continue
                    yield Message(
                        id=msg_id,
                        channel=self.channel,
                        sender=user,
                        content=msg.get("text", ""),
                        metadata={"slack": msg},
                    )
            except Exception as exc:  # pragma: no cover - network resilience
                logger.warning("SlackAdapter.receive polling error: %s", exc)
            await asyncio.sleep(5)

enqueue(message: Message) -> None

Push an inbound :class:Message onto the internal queue.

Used by the Slack webhook router to feed real message.changed events into the adapter; consumers can drain them via :meth:receive (which is overridden below to prefer the queue).

Source code in src/ciel/gateway/adapter_slack.py 
71
72
73
74
75
76
77
78
def enqueue(self, message: Message) -> None:
    """Push an inbound :class:`Message` onto the internal queue.

    Used by the Slack webhook router to feed real ``message.changed``
    events into the adapter; consumers can drain them via
    :meth:`receive` (which is overridden below to prefer the queue).
    """
    self._inbound.put_nowait(message)

receive() -> AsyncIterator[Message] async

Yield inbound Slack messages via a simple polling loop.

This is a lenient, best-effort implementation: it relies on Slack's conversations.history to pull recent messages and yields any that are not authored by the bot user. It polls every few seconds.

The primary, fully-tested contract of this adapter is :meth:send. The receive loop is provided so the adapter is genuinely bidirectional, but it is intentionally simple and skips advanced cursor pagination and deduplication. Inbound event handling is better served by mounting the Slack webhook router (see :func:create_slack_webhook_router) onto the FastAPI app, which enqueues real message.changed events.

The loop runs forever; the caller is responsible for cancellation.

Source code in src/ciel/gateway/adapter_slack.py 
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
181
182
183
184
185
186
187
async def receive(self) -> AsyncIterator[Message]:
    """Yield inbound Slack messages via a simple polling loop.

    This is a lenient, best-effort implementation: it relies on Slack's
    ``conversations.history`` to pull recent messages and yields any that
    are not authored by the bot user. It polls every few seconds.

    The primary, fully-tested contract of this adapter is :meth:`send`.
    The receive loop is provided so the adapter is genuinely bidirectional,
    but it is intentionally simple and skips advanced cursor pagination and
    deduplication. Inbound event handling is better served by mounting the
    Slack webhook router (see :func:`create_slack_webhook_router`) onto the
    FastAPI app, which enqueues real ``message.changed`` events.

    The loop runs forever; the caller is responsible for cancellation.
    """
    client = self._get_client()
    if not self.channel:
        logger.warning(
            "SlackAdapter.receive: no default channel set; polling skipped."
        )
        return
    seen: set[str] = set()
    loop = asyncio.get_running_loop()
    while True:
        try:
            response = await loop.run_in_executor(
                None,
                lambda: client.conversations_history(channel=self.channel, limit=20),
            )
            for msg in response.get("messages", []):
                msg_id = msg.get("ts")
                user = msg.get("user")
                if msg_id in seen:
                    continue
                seen.add(msg_id)
                if self.bot_user_id and user == self.bot_user_id:
                    continue
                yield Message(
                    id=msg_id,
                    channel=self.channel,
                    sender=user,
                    content=msg.get("text", ""),
                    metadata={"slack": msg},
                )
        except Exception as exc:  # pragma: no cover - network resilience
            logger.warning("SlackAdapter.receive polling error: %s", exc)
        await asyncio.sleep(5)

send(message: Message) -> None async

Send message.content to Slack via chat.postMessage.

The destination channel is resolved from message.metadata['channel'] falling back to the adapter's default channel. A clear :class:ValueError is raised when no channel can be resolved.

Network access is required for this call to succeed against the real Slack API; tests pass a mocked WebClient to verify behaviour without touching the network.

Parameters:

Name Type Description Default
message Message

The :class:Message to deliver.

 required
Source code in src/ciel/gateway/adapter_slack.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
async def send(self, message: Message) -> None:
    """Send ``message.content`` to Slack via ``chat.postMessage``.

    The destination channel is resolved from ``message.metadata['channel']``
    falling back to the adapter's default ``channel``. A clear
    :class:`ValueError` is raised when no channel can be resolved.

    Network access is required for this call to succeed against the real
    Slack API; tests pass a mocked ``WebClient`` to verify behaviour
    without touching the network.

    Args:
        message: The :class:`Message` to deliver.
    """
    client = self._get_client()
    channel = self._resolve_channel(message)
    # chat_postMessage is synchronous in slack_sdk; run it in a thread so
    # this stays friendly to the asyncio event loop.
    loop = asyncio.get_running_loop()
    await loop.run_in_executor(
        None,
        lambda: client.chat_postMessage(channel=channel, text=message.text()),
    )
    logger.debug("SlackAdapter.send -> channel=%s len=%d", channel, len(message.text()))

Webhook routers para adapters de canal (Fase 8 — Teams/Discord/Web UI).

Estos routers montan endpoints HTTP que convierten eventos entrantes de cada canal en ciel.gateway.adapter.Message y los empujan (enqueue) al adapter correspondiente, de forma que el runtime puede consumirlos vía adapter.receive(). Siguen el mismo patrón que ciel.gateway.create_slack_webhook_router.

OFFLINE-SAFE: los routers no requieren red para importarse ni para arrancar; solo la entrega real de send (en los adapters) toca la red.

logger = logging.getLogger(__name__) module-attribute

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)

create_discord_webhook_router(adapter, *, path: str = '/v1/messaging/discord') -> APIRouter

Router que ingiere eventos MESSAGE_CREATE de Discord al adapter.

Discord entrega un JSON con content, author/id y channel_id. El router lo convierte en Message y lo enqueuea.

Source code in src/ciel/gateway/messaging.py 
 72
 73
 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
def create_discord_webhook_router(
    adapter,
    *,
    path: str = "/v1/messaging/discord",
) -> APIRouter:
    """Router que ingiere eventos ``MESSAGE_CREATE`` de Discord al ``adapter``.

    Discord entrega un JSON con ``content``, ``author``/``id`` y ``channel_id``.
    El router lo convierte en ``Message`` y lo enqueuea.
    """
    router = APIRouter()

    @router.post(path)
    async def discord_events(request: Request):
        try:
            payload = await request.json()
        except Exception as exc:
            return {"status": "rejected", "error": f"invalid json: {exc}"}

        if payload.get("type") == 1:
            # Discord Ping (interacción): responde con el mismo id.
            return {"type": 1}

        content = payload.get("content", "")
        if not content:
            return {"status": "ok"}  # eventos sin contenido (presence, etc.)
        author = payload.get("author", {})
        sender = author.get("username") or author.get("id")
        message = Message(
            id=payload.get("id") or str(__import__("uuid").uuid4()),
            channel="discord",
            sender=sender,
            content=content,
            metadata={"discord_event": payload, "channel_id": payload.get("channel_id")},
        )
        enqueue = getattr(adapter, "enqueue", None)
        if callable(enqueue):
            enqueue(message)
        else:  # pragma: no cover - adapter sin enqueue
            logger.warning("DiscordAdapter has no enqueue(); dropping %s", message.id)
        return {"status": "accepted", "id": message.id}

    @router.get(f"{path}/health")
    async def discord_health():
        return {"status": "ok", "channel": "discord"}

    return router

create_teams_webhook_router(adapter, *, path: str = '/v1/messaging/teams', signing_secret: Optional[str] = None) -> APIRouter

Router que ingiere eventos de Microsoft Teams al adapter.

Teams entrega cargas JSON con text (y opcionalmente from / channelData). El router los convierte en Message y los enqueuea. Si se pasa signing_secret se puede validar la firma HMAC (fuera de alcance en smoke tests); sin él, se acepta el payload tal cual.

Source code in src/ciel/gateway/messaging.py 
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
64
65
66
67
68
69
def create_teams_webhook_router(
    adapter,
    *,
    path: str = "/v1/messaging/teams",
    signing_secret: Optional[str] = None,
) -> APIRouter:
    """Router que ingiere eventos de Microsoft Teams al ``adapter``.

    Teams entrega cargas JSON con ``text`` (y opcionalmente ``from`` /
    ``channelData``). El router los convierte en ``Message`` y los enqueuea.
    Si se pasa ``signing_secret`` se puede validar la firma HMAC (fuera de
    alcance en smoke tests); sin él, se acepta el payload tal cual.
    """
    router = APIRouter()

    @router.post(path)
    async def teams_events(request: Request):
        try:
            payload = await request.json()
        except Exception as exc:
            return {"status": "rejected", "error": f"invalid json: {exc}"}

        text = payload.get("text") or payload.get("body") or ""
        if not text:
            return {"status": "rejected", "error": "missing text"}
        sender = payload.get("from", {}).get("id") if isinstance(payload.get("from"), dict) else payload.get("from")
        message = Message(
            id=payload.get("id") or str(__import__("uuid").uuid4()),
            channel="teams",
            sender=sender,
            content=text,
            metadata={"teams_event": payload},
        )
        enqueue = getattr(adapter, "enqueue", None)
        if callable(enqueue):
            enqueue(message)
        else:  # pragma: no cover - adapter sin enqueue
            logger.warning("TeamsAdapter has no enqueue(); dropping %s", message.id)
        return {"status": "accepted", "id": message.id}

    @router.get(f"{path}/health")
    async def teams_health():
        return {"status": "ok", "channel": "teams"}

    return router

create_webui_router(adapter, *, path: str = '/v1/messaging/webui') -> APIRouter

Router de Web UI: POST para enviar mensajes al runtime, GET para sondear.

La UI hace POST {path} con un Message y lo enqueuea en el WebUIAdapter; GET {path}/outbound sondea el siguiente mensaje saliente del adapter (polling).

Source code in src/ciel/gateway/messaging.py 
121
122
123
124
125
126
127
128
129
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
def create_webui_router(
    adapter,
    *,
    path: str = "/v1/messaging/webui",
) -> APIRouter:
    """Router de Web UI: POST para enviar mensajes al runtime, GET para sondear.

    La UI hace ``POST {path}`` con un ``Message`` y lo enqueuea en el
    ``WebUIAdapter``; ``GET {path}/outbound`` sondea el siguiente mensaje
    saliente del adapter (polling).
    """
    from ciel.adapters import WebUIAdapter

    router = APIRouter()

    @router.post(path)
    async def webui_inbound(request: Request):
        try:
            payload = await request.json()
        except Exception as exc:
            return {"status": "rejected", "error": f"invalid json: {exc}"}
        content = payload.get("content")
        if not content:
            return {"status": "rejected", "error": "missing content"}
        message = Message(
            id=payload.get("id") or str(__import__("uuid").uuid4()),
            channel="webui",
            sender=payload.get("sender"),
            content=content,
            metadata=payload.get("metadata") or {},
        )
        enqueue = getattr(adapter, "enqueue", None)
        if callable(enqueue):
            enqueue(message)
        else:  # pragma: no cover - adapter sin enqueue
            logger.warning("WebUIAdapter has no enqueue(); dropping %s", message.id)
        return {"status": "accepted", "id": message.id}

    @router.get(f"{path}/outbound")
    async def webui_outbound():
        # Polling: devuelve el siguiente mensaje saliente o 204 si no hay.
        if not isinstance(adapter, WebUIAdapter):
            return {"status": "ok", "channel": "webui"}
        try:
            message = adapter._outbound.get_nowait()
        except Exception:
            return {"status": "no_content"}
        return {
            "status": "ok",
            "id": message.id,
            "content": message.text(),
            "sender": message.sender,
            "metadata": message.metadata,
        }

    @router.get(f"{path}/health")
    async def webui_health():
        return {"status": "ok", "channel": "webui"}

    return router

__all__ = ['MCPClient', 'MCPHTTPTransport', 'MCPStdioTransport', 'MCPTransport', 'MCPServer', 'DefaultAgentRuntimeMCPHost', 'MCPHostToolProvider'] module-attribute

DefaultAgentRuntimeMCPHost

Host DefaultAgentRuntime behavior through an MCP server boundary.

Source code in src/ciel/gateway/mcp/integration.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
class DefaultAgentRuntimeMCPHost:
    """Host DefaultAgentRuntime behavior through an MCP server boundary."""

    def __init__(
        self,
        runtime: Any,
        *,
        audit_sink: Optional[Any] = None,
        tenant_id: Optional[str] = None,
        agent: str = "default",
    ) -> None:
        self.runtime = runtime
        self.audit_sink = audit_sink or NullAuditSink()
        self.tenant_id = tenant_id
        self.agent = agent
        self.server = MCPServer(audit_sink=self.audit_sink, tenant_id=tenant_id)

    async def _emit(self, event: AuditEvent, *, tenant_id: Optional[str] = None) -> AuditEvent:
        normalized = propagate(event, tenant_id=tenant_id)
        await self.audit_sink.write(normalized)
        return normalized

    async def handle_request(self, payload: Mapping[str, Any]) -> Dict[str, Any]:
        response = await self.server.handle(payload)
        await self._emit(
            AuditEvent(
                event="mcp.host.request",
                agent=self.agent,
                data={"method": payload.get("method"), "request_id": payload.get("id")},
            ),
            tenant_id=self.tenant_id,
        )
        return response

MCPClient

MCP protocol client over an explicit transport tenant-aware runtime.

Source code in src/ciel/gateway/mcp/client.py 
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
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
class MCPClient:
    """MCP protocol client over an explicit transport tenant-aware runtime."""

    def __init__(
        self,
        *,
        transport: MCPTransport,
        tenant_id: Optional[str] = None,
        audit_sink: Optional[Any] = None,
        agent: str = "default",
    ) -> None:
        if audit_sink is None:
            audit_sink = NullAuditSink()
        self.transport = transport
        self.tenant_id = tenant_id
        self.audit_sink = audit_sink
        self.agent = agent
        self.server_version: Optional[str] = None

    def _request_id(self) -> str:
        return str(uuid.uuid4())

    async def _send(self, payload: Dict[str, Any]) -> None:
        await self.transport.send(payload)

    async def _receive(self) -> AsyncIterator[Dict[str, Any]]:
        async for message in self.transport.receive():
            yield message

    async def _emit(self, event: AuditEvent) -> AuditEvent:
        normalized = propagate(event, tenant_id=self.tenant_id)
        await self.audit_sink.write(normalized)
        return normalized

    async def initialize(
        self,
        *,
        client_name: str = "ciel-mcp-client",
        client_version: str = "0.1.0",
    ) -> None:
        payload = {
            "jsonrpc": "2.0",
            "id": self._request_id(),
            "method": "initialize",
            "params": {
                "clientName": client_name,
                "clientVersion": client_version,
                "tenant_id": self.tenant_id,
            },
        }
        await self._send(payload)
        async for message in self._receive():
            result = message.get("result")
            if result and isinstance(result, dict):
                self.server_version = result.get("serverVersion") or result.get("version")
            break
        await self._emit(
            AuditEvent(
                event="mcp.client.initialize",
                agent=self.agent,
                data={"client": client_name, "clientVersion": client_version},
            )
        )

MCPHTTPTransport

Bases: MCPTransport

Source code in src/ciel/gateway/mcp/transports.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
class MCPHTTPTransport(MCPTransport):
    def __init__(self, *, base_url: str, client: Any) -> None:
        self._base_url = base_url.rstrip("/")
        self._client = client

    async def send(self, message: Mapping[str, Any]) -> None:
        await self._client.post(f"{self._base_url}/mcp", json=message)

    async def receive(self) -> AsyncIterator[Mapping[str, Any]]:
        response = await self._client.get(f"{self._base_url}/mcp/events")
        async for line in response.aiter_lines():
            line = line.strip()
            if not line:
                continue
            try:
                yield json.loads(line)
            except json.JSONDecodeError:
                continue

    async def close(self) -> None:
        await self._client.aclose()

MCPHostToolProvider

Bases: ToolProvider

Expose an MCP server as a Ciel ToolProvider.

Source code in src/ciel/gateway/mcp/integration.py 
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
64
65
66
67
68
69
70
71
class MCPHostToolProvider(ToolProvider):
    """Expose an MCP server as a Ciel ToolProvider."""

    def __init__(self, server: Any, *, default_toolset: str = "default") -> None:
        self.server = server
        self.default_toolset = default_toolset

    async def tool_specs(self, toolset: str) -> Sequence[ToolSpec]:
        payload = await self.server.handle(
            {"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {"toolset": toolset or self.default_toolset}}
        )
        result = payload.get("result") or {}
        specs = []
        for spec in result.get("tools", []):
            specs.append(
                ToolSpec(
                    name=spec.get("name", ""),
                    description=spec.get("description", ""),
                    parameters=spec.get("parameters", {}),
                    strict=spec.get("strict", False),
                    metadata=spec.get("metadata", {}),
                )
            )
        return specs

    async def execute(
        self,
        *,
        tenant_id: Optional[str] = None,
        toolset: str,
        name: str,
        arguments: Dict[str, Any],
        tool_call_id: str,
    ) -> ToolResult:
        payload = await self.server.handle(
            {
                "jsonrpc": "2.0",
                "id": tool_call_id,
                "method": "tools/call",
                "params": {
                    "id": tool_call_id,
                    "tenant_id": tenant_id,
                    "name": name,
                    "arguments": arguments,
                    "toolset": toolset or self.default_toolset,
                },
            }
        )
        result = payload.get("result") or {}
        return ToolResult(
            id=result.get("id", tool_call_id),
            name=name,
            output=result.get("output"),
            error=(result.get("error") or {}).get("message") if isinstance(result.get("error"), dict) else result.get("error"),
            metadata=result.get("metadata", {}),
        )

MCPServer

Minimal MCP-style JSON-RPC endpoint with tenant-aware handlers.

Source code in src/ciel/gateway/mcp/server.py 
 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
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 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
class MCPServer:
    """Minimal MCP-style JSON-RPC endpoint with tenant-aware handlers."""

    def __init__(
        self,
        *,
        provider: Optional[ToolProvider] = None,
        tenant_id: Optional[str] = None,
        audit_sink: Optional[Any] = None,
    ) -> None:
        self.provider = provider
        self.tenant_id = tenant_id
        self.audit_sink = audit_sink or NullAuditSink()
        self.handlers: Dict[str, Any] = {
            "initialize": self.handle_initialize,
            "shutdown": self.handle_shutdown,
            "tools/list": self.handle_tools_list,
            "tools/call": self.handle_tools_call,
        }

    async def handle(self, payload: Mapping[str, Any]) -> Dict[str, Any]:
        method = payload.get("method")
        request_id = payload.get("id")
        if method not in self.handlers:
            return self._error(request_id, -32601, "Method not found")
        try:
            result = await self.handlers[method](payload.get("params", {}))
        except Exception as exc:  # pragma: no cover - defensive path
            logger.exception("MCP handler error")
            return self._error(request_id, -32603, str(exc))
        return {"jsonrpc": "2.0", "id": request_id, "result": result}

    async def handle_initialize(self, params: Dict[str, Any]) -> Dict[str, Any]:
        tenant_id = params.get("tenant_id") or self.tenant_id
        self.tenant_id = tenant_id
        event = AuditEvent(
            event="mcp.server.initialize",
            data={
                "clientName": params.get("clientName"),
                "clientVersion": params.get("clientVersion"),
            },
        )
        normalized = propagate(event, tenant_id=tenant_id)
        await self.audit_sink.write(normalized)
        return {"status": "initialized", "version": "0.1.0", "tenant_id": tenant_id}

    async def handle_shutdown(self, params: Dict[str, Any]) -> Dict[str, Any]:
        event = AuditEvent(event="mcp.server.shutdown")
        normalized = propagate(event, tenant_id=self.tenant_id)
        await self.audit_sink.write(normalized)
        return {"status": "shutdown"}

    async def handle_tools_list(self, params: Dict[str, Any]) -> Dict[str, Any]:
        toolset = params.get("toolset") or "default"
        tenant_id = params.get("tenant_id") or self.tenant_id
        specs: Sequence[ToolSpec] = ()
        if self.provider is not None:
            specs = await self.provider.tool_specs(tenant_id, toolset)
        return {"tools": [self._serialize_spec(spec) for spec in specs]}

    async def handle_tools_call(self, params: Dict[str, Any]) -> Dict[str, Any]:
        if self.provider is None:
            return {"error": {"code": -32000, "message": "no tool provider configured"}}
        toolset = params.get("toolset") or "default"
        tenant_id = params.get("tenant_id") or self.tenant_id
        call_id = params.get("id") or params.get("tool_call_id") or str(uuid.uuid4())
        result = await self.provider.execute(
            tenant_id=tenant_id,
            toolset=toolset,
            name=params["name"],
            arguments=params.get("arguments") or {},
            tool_call_id=call_id,
        )
        response: Dict[str, Any] = {
            "id": result.id,
            "name": result.name,
            "output": result.output,
            "metadata": dict(result.metadata),
        }
        if result.error:
            response["error"] = result.error
        return response

    def _serialize_spec(self, spec: ToolSpec) -> Dict[str, Any]:
        payload: Dict[str, Any] = {
            "name": spec.name,
            "description": spec.description,
            "parameters": dict(spec.parameters),
        }
        if spec.strict:
            payload["strict"] = spec.strict
        if spec.metadata:
            payload["metadata"] = dict(spec.metadata)
        return payload

    @staticmethod
    def _error(request_id: Optional[Any], code: int, message: str) -> Dict[str, Any]:
        return {
            "jsonrpc": "2.0",
            "id": request_id,
            "error": {"code": code, "message": message},
        }

MCPStdioTransport

Bases: MCPTransport

Source code in src/ciel/gateway/mcp/transports.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
class MCPStdioTransport(MCPTransport):
    def __init__(self, *, stdout_write: Any, stdin_readline: Any) -> None:
        self._stdout = stdout_write
        self._stdin = stdin_readline

    async def send(self, message: Mapping[str, Any]) -> None:
        self._stdout.write(json.dumps(message) + "\n")
        self._stdout.flush()

    async def receive(self) -> AsyncIterator[Mapping[str, Any]]:
        while True:
            line = await self._stdin()
            if not line:
                break
            line = line.strip()
            if not line:
                continue
            try:
                yield json.loads(line)
            except json.JSONDecodeError:
                continue

    async def close(self) -> None:
        return None

MCPTransport

Bases: ABC

Source code in src/ciel/gateway/mcp/transports.py 
 9
10
11
12
13
14
15
16
17
18
19
20
class MCPTransport(ABC):
    @abstractmethod
    async def send(self, message: Mapping[str, Any]) -> None:
        raise NotImplementedError

    @abstractmethod
    async def receive(self) -> AsyncIterator[Mapping[str, Any]]:
        raise NotImplementedError

    @abstractmethod
    async def close(self) -> None:
        raise NotImplementedError

Optional transport-level API key authentication for gateway surfaces.

This module provides a FastAPI dependency that enforces an optional API key on protected routes. The behaviour is driven by the CIEL_API_KEY environment variable (or an explicit key passed by the caller):

  • When a key is configured, every protected route requires a valid key supplied via the Authorization: Bearer <key> header or the X-API-Key header. A missing or mismatched key yields 401 Unauthorized.
  • When no key is configured (the default for offline/dev/smoke-test deployments) the dependency is a no-op and the request proceeds. This keeps the gateway bootable with zero configuration.

The comparison uses :func:hmac.compare_digest to avoid timing side-channels.

API_KEY_ENV = 'CIEL_API_KEY' module-attribute

__all__ = ['API_KEY_ENV', 'read_expected_key', 'make_auth_dependency', 'require_api_key', 'depends', 'AuthContext', 'make_oidc_dependency'] module-attribute

require_api_key = make_auth_dependency(expected_key=None) module-attribute

AuthContext dataclass

Contexto de autenticación resuelto por la dependency.

subject y role provienen de los claims OIDC cuando el token se validó contra un IdP real; en modo api_key/open ambos son None.

Source code in src/ciel/gateway/auth.py 
121
122
123
124
125
126
127
128
129
130
131
132
@dataclass
class AuthContext:
    """Contexto de autenticación resuelto por la dependency.

    ``subject`` y ``role`` provienen de los claims OIDC cuando el token se validó
    contra un IdP real; en modo api_key/open ambos son ``None``.
    """

    subject: Optional[str] = None
    role: Optional[str] = None
    claims: Optional[dict] = None
    via: str = "open"  # "open" | "api_key" | "oidc"

_extract_provided_key(authorization: Optional[str], x_api_key: Optional[str]) -> Optional[str]

Pull the key out of the supported headers.

Source code in src/ciel/gateway/auth.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def _extract_provided_key(
    authorization: Optional[str],
    x_api_key: Optional[str],
) -> Optional[str]:
    """Pull the key out of the supported headers."""
    if authorization:
        value = authorization.strip()
        # Accept both "Bearer <key>" and a bare key.
        if value.lower().startswith("bearer "):
            return value[len("bearer ") :].strip() or None
        return value or None
    if x_api_key:
        return x_api_key.strip() or None
    return None

_reject() -> HTTPException

Source code in src/ciel/gateway/auth.py 
71
72
73
74
75
76
def _reject() -> HTTPException:
    return HTTPException(
        status_code=401,
        detail="Invalid or missing API key.",
        headers={"WWW-Authenticate": "Bearer"},
    )

depends(api_key: Optional[str] = None)

Helper returning a Depends(...) for a given key/configuration.

Source code in src/ciel/gateway/auth.py 
109
110
111
def depends(api_key: Optional[str] = None):
    """Helper returning a ``Depends(...)`` for a given key/configuration."""
    return Depends(make_auth_dependency(expected_key=api_key))

make_auth_dependency(expected_key: Optional[str] = None)

Build a FastAPI dependency enforcing an optional API key.

The returned callable reads expected_key (explicit) or falls back to the CIEL_API_KEY environment variable at request time. This lets the same closure be reused across requests while still honoring runtime env changes (e.g. monkeypatch.setenv in tests).

Source code in src/ciel/gateway/auth.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def make_auth_dependency(expected_key: Optional[str] = None):
    """Build a FastAPI dependency enforcing an optional API key.

    The returned callable reads ``expected_key`` (explicit) or falls back to
    the ``CIEL_API_KEY`` environment variable at request time. This lets the
    same closure be reused across requests while still honoring runtime env
    changes (e.g. ``monkeypatch.setenv`` in tests).
    """

    async def _guard(
        authorization: Optional[str] = Header(default=None, alias="Authorization"),
        x_api_key: Optional[str] = Header(default=None, alias="X-API-Key"),
    ) -> None:
        expected = read_expected_key(explicit=expected_key)
        if expected is None:
            # Open mode: no transport auth configured.
            return
        provided = _extract_provided_key(authorization, x_api_key)
        if provided is None or not hmac.compare_digest(provided, expected):
            raise _reject()

    return _guard

make_oidc_dependency(*, verifier=None, enabled: Optional[bool] = None, api_key: Optional[str] = None)

Build a FastAPI dependency that enforces OIDC when enabled.

Behaviour:

  • When OIDC is disabled (enabled is False, or None and CIEL_OIDC_ENABLED is unset), it fully delegates to the api_key guard (retrocompatibilidad total, open-mode por defecto).
  • When OIDC is enabled, every protected route requires a valid Bearer JWT. The token is verified (local key or JWKS) and its claims are mapped to an RBAC role. Missing/invalid token => 401.

The dependency returns an :class:AuthContext.

Source code in src/ciel/gateway/auth.py 
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
181
182
183
184
185
186
187
188
189
def make_oidc_dependency(
    *,
    verifier=None,
    enabled: Optional[bool] = None,
    api_key: Optional[str] = None,
):
    """Build a FastAPI dependency that enforces OIDC when enabled.

    Behaviour:

    * When OIDC is **disabled** (``enabled`` is False, or None and
      ``CIEL_OIDC_ENABLED`` is unset), it fully delegates to the api_key guard
      (retrocompatibilidad total, open-mode por defecto).
    * When OIDC is **enabled**, every protected route requires a valid Bearer
      JWT. The token is verified (local key or JWKS) and its claims are mapped to
      an RBAC role. Missing/invalid token => ``401``.

    The dependency returns an :class:`AuthContext`.
    """
    from ciel.enterprise.rbac import FeatureUnavailable, OIDCVerifier

    def _is_enabled() -> bool:
        if enabled is not None:
            return enabled
        return OIDCVerifier.enabled_from_env()

    api_key_guard = make_auth_dependency(expected_key=api_key)

    async def _guard(
        authorization: Optional[str] = Header(default=None, alias="Authorization"),
        x_api_key: Optional[str] = Header(default=None, alias="X-API-Key"),
    ) -> AuthContext:
        if not _is_enabled():
            # Fallback: comportamiento api_key (open-mode si no hay clave).
            await api_key_guard(authorization=authorization, x_api_key=x_api_key)
            return AuthContext(via="open" if read_expected_key(explicit=api_key) is None else "api_key")

        token = _extract_provided_key(authorization, x_api_key)
        if token is None:
            raise _reject()
        active_verifier = verifier or OIDCVerifier.from_config()
        try:
            claims, role = active_verifier.verify_and_map_role(token)
        except FeatureUnavailable as exc:
            raise HTTPException(status_code=503, detail=str(exc)) from exc
        except Exception as exc:  # verificación fallida
            raise HTTPException(
                status_code=401,
                detail="Invalid OIDC token.",
                headers={"WWW-Authenticate": "Bearer"},
            ) from exc
        subject = claims.get("sub") if isinstance(claims, dict) else None
        return AuthContext(subject=subject, role=role, claims=claims, via="oidc")

    return _guard

read_expected_key(*, explicit: Optional[str] = None) -> Optional[str]

Return the configured API key, preferring an explicit value.

Parameters

explicit: A key passed directly by the caller. If provided (even when empty string), it takes precedence over the environment.

Source code in src/ciel/gateway/auth.py 
41
42
43
44
45
46
47
48
49
50
51
52
def read_expected_key(*, explicit: Optional[str] = None) -> Optional[str]:
    """Return the configured API key, preferring an explicit value.

    Parameters
    ----------
    explicit:
        A key passed directly by the caller. If provided (even when empty
        string), it takes precedence over the environment.
    """
    if explicit is not None:
        return explicit or None
    return os.getenv(API_KEY_ENV) or None