Skip to content

The served server

octomate.mcp.server

Compose gateway, history, and user-installed MCP tools under one server.

Runtimes use the same discovery and call helpers for every installed MCP. Inkling already has gateway and history tools in process, so it mounts only tentacles_mcp; the other runtimes mount the combined server.

OCTOMATE_SERVER_NAME module-attribute 

OCTOMATE_SERVER_NAME = 'octomate'

OCTOMATE_MCP_PATH module-attribute 

OCTOMATE_MCP_PATH = f'/{OCTOMATE_SERVER_NAME}/mcp'

HISTORY_NAMESPACE module-attribute 

HISTORY_NAMESPACE = 'history'

TENTACLES_SERVER_NAME module-attribute 

TENTACLES_SERVER_NAME = 'tentacles'

LIST_MCP_TOOLS module-attribute 

LIST_MCP_TOOLS = 'mcp_list_tools'

CALL_MCP_TOOL module-attribute 

CALL_MCP_TOOL = 'mcp_call_tool'

LIST_MCPS module-attribute 

LIST_MCPS = 'mcp_list_servers'

LIST_MCP_TENTACLES module-attribute 

LIST_MCP_TENTACLES = 'mcp_list_tentacles'

INSTALL_MCP module-attribute 

INSTALL_MCP = 'mcp_install'

UNINSTALL_MCP module-attribute 

UNINSTALL_MCP = 'mcp_uninstall'

ENABLE_MCP module-attribute 

ENABLE_MCP = 'mcp_enable'

DISABLE_MCP module-attribute 

DISABLE_MCP = 'mcp_disable'

history_tool 

history_tool(name: str) -> str

A history tool's served name, from the capability's own.

Source code in octomate/mcp/server.py 
66
67
68
def history_tool(name: str) -> str:
    """A history tool's served name, from the capability's own."""
    return f"{HISTORY_NAMESPACE}_{HISTORY_TOOL_NAMES[name]}"

tentacle_instructions 

tentacle_instructions() -> str
Source code in octomate/mcp/server.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def tentacle_instructions() -> str:
    return (
        "When searching for tools, also check the MCP servers proxied by Octomate "
        "before concluding that a needed tool is unavailable. "
        f"Call `{LIST_MCPS}` to discover the current user's installed MCPs. "
        f"Use `{LIST_MCP_TOOLS}` with an enabled namespace, then `{CALL_MCP_TOOL}` "
        "with the exact discovered tool name and arguments. "
        f"When the user asks to install an MCP, use `{LIST_MCP_TENTACLES}` to find "
        f"configured offerings, then `{INSTALL_MCP}`. Remote MCPs can also be "
        "installed by HTTPS URL with no authentication or dynamic OAuth. "
        f"Use `{ENABLE_MCP}`, `{DISABLE_MCP}`, and `{UNINSTALL_MCP}` with an "
        "installation's id when the user requests those changes. "
        "If authorization is required, use `oauth_connect` with that namespace "
        "and an available OAuth flow to send the user a private authorization "
        "link, then `oauth_confirm`. Installing does not authorize an MCP. "
        "OAuth links require a channel with private delivery. Enter personal "
        "bearer tokens through the authenticated MCP HTTP API, never tool arguments. "
        "When an unregistered channel user asks to link the profile driving the "
        "current turn to an Octomate account, use `oauth_link_profile`; it privately "
        "delivers the browser flow without accepting or returning an identity."
    )

octomate_instructions 

octomate_instructions() -> str
Source code in octomate/mcp/server.py 
 94
 95
 96
 97
 98
 99
100
101
def octomate_instructions() -> str:
    return "\n".join(
        [
            gateway_instructions(gateway_tool),
            history_instructions(history_tool),
            tentacle_instructions(),
        ]
    )

tentacles_mcp 

tentacles_mcp(
    resolve_session: Callable[
        [], Awaitable[OctomateSession]
    ],
    *,
    manager: McpManager,
) -> FastMCP

Serve discovery, calls, and authorization for the current user's MCPs.

Source code in octomate/mcp/server.py 
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
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
def tentacles_mcp(
    resolve_session: Callable[[], Awaitable[OctomateSession]],
    *,
    manager: McpManager,
) -> FastMCP:
    """Serve discovery, calls, and authorization for the current user's MCPs."""
    mcp = FastMCP(
        name=TENTACLES_SERVER_NAME,
        instructions=tentacle_instructions(),
    )
    oauth = FastMCP(OAUTH_NAMESPACE)
    session = Depends(resolve_session)
    mount_oauth(oauth, session, manager=manager)
    mcp.mount(oauth, namespace=OAUTH_NAMESPACE)

    async def require_user(
        scope: OctomateSession = session,
    ) -> User:
        if scope.user_profile is not None:
            user = await manager.users.owner(scope.user_profile)
            if user is not None:
                return user
        raise ToolError("MCP management requires a registered user")

    @mcp.tool(
        name=LIST_MCP_TENTACLES,
        description="List configured MCP tentacles the current user can install.",
        annotations={"readOnlyHint": True},
    )
    async def list_tentacles(
        # Depends declares injection; FastMCP resolves it separately for each call.
        user: User = Depends(require_user),  # noqa: B008
    ) -> list[McpTentacleInfo]:
        return manager.available()

    @mcp.tool(
        name=LIST_MCPS,
        description=(
            "List the current user's installed MCPs, including disabled ones, "
            "stored OAuth status, and available authorization flows. Does not poll providers."
        ),
        annotations={"readOnlyHint": True},
    )
    async def list_servers() -> list[McpServerSummary]:
        scope = await resolve_session()
        if scope.user_profile is None:
            return []
        owner = await manager.users.owner(scope.user_profile)
        if owner is None:
            return []
        return [
            await manager.summary(owner, instance)
            for instance in await manager.list(user_id=owner.id)
        ]

    @mcp.tool(
        name=INSTALL_MCP,
        description=(
            "Install an MCP for the current user. For a configured tentacle, pass its "
            "id and URL from mcp_list_tentacles and omit auth. For a remote URL, "
            "choose none or oauth. Namespace is a unique local name without personal/. "
            "Installation enables the MCP; OAuth authorization is a separate step."
        ),
        annotations={"destructiveHint": False},
    )
    async def install(
        name: str,
        namespace: str,
        url: HttpsUrl,
        *,
        tentacle_id: str | None = None,
        auth: Literal["none", "oauth"] | None = None,
        user: User = Depends(require_user),  # noqa: B008
    ) -> McpServerSummary:
        match auth:
            case "none":
                authentication = NoAuth()
            case "oauth":
                authentication = OAuth()
            case None:
                authentication = None
        try:
            instance = await manager.install(
                user_id=user.id,
                request=McpInstallRequest(
                    name=name,
                    namespace=namespace,
                    url=url,
                    tentacle_id=tentacle_id,
                    auth=authentication,
                ),
            )
        except (McpUnavailable, ValueError) as error:
            raise ToolError(str(error)) from error
        return await manager.summary(user, instance)

    @mcp.tool(
        name=ENABLE_MCP,
        description="Enable one of the current user's installed MCPs, keeping its saved authorization.",
        annotations={"destructiveHint": False, "idempotentHint": True},
    )
    async def enable(
        # OpenAI strict mode needs the standard format; UUID7 still validates values.
        mcp_id: Annotated[UUID7, WithJsonSchema({"type": "string", "format": "uuid"})],
        user: User = Depends(require_user),  # noqa: B008
    ) -> McpServerSummary:
        try:
            instance = await manager.enable(user_id=user.id, mcp_id=mcp_id)
        except McpUnavailable as error:
            raise ToolError(str(error)) from error
        return await manager.summary(user, instance)

    @mcp.tool(
        name=DISABLE_MCP,
        description=(
            "Disable one of the current user's MCPs and close its cached client "
            "immediately. Keeps the installation and saved authorization."
        ),
        annotations={"destructiveHint": False, "idempotentHint": True},
    )
    async def disable(
        mcp_id: Annotated[UUID7, WithJsonSchema({"type": "string", "format": "uuid"})],
        user: User = Depends(require_user),  # noqa: B008
    ) -> McpServerSummary:
        try:
            instance = await manager.disable(user_id=user.id, mcp_id=mcp_id)
        except McpUnavailable as error:
            raise ToolError(str(error)) from error
        return await manager.summary(user, instance)

    @mcp.tool(
        name=UNINSTALL_MCP,
        description=(
            "Remove one of the current user's MCP installations and its saved "
            "credentials, closing its cached client immediately. Does not revoke "
            "authorization at the provider."
        ),
        annotations={"destructiveHint": True},
    )
    async def uninstall(
        mcp_id: Annotated[UUID7, WithJsonSchema({"type": "string", "format": "uuid"})],
        user: User = Depends(require_user),  # noqa: B008
    ) -> str:
        try:
            await manager.uninstall(user_id=user.id, mcp_id=mcp_id)
        except McpUnavailable as error:
            raise ToolError(str(error)) from error
        return "MCP uninstalled."

    @mcp.tool(
        name=LIST_MCP_TOOLS,
        description="Load tool schemas for one of the current user's installed MCPs.",
        annotations={"readOnlyHint": True},
    )
    async def list_tools(namespace: str) -> McpToolCatalog:
        try:
            return await manager.catalog(
                await resolve_session(), namespace, refresh=True
            )
        except McpUnavailable as error:
            raise ToolError(str(error)) from error
        except (httpx2.HTTPError, MCPError, TimeoutError) as error:
            raise ToolError("MCP upstream request failed") from error

    @mcp.tool(
        name=CALL_MCP_TOOL,
        description=(
            f"Call a tool discovered with `{LIST_MCP_TOOLS}` as the current caller. "
        ),
    )
    async def call_tool(
        namespace: str,
        name: Annotated[
            str, Field(description="The exact tool name returned by discovery.")
        ],
        arguments: dict[str, JsonValue],
    ) -> ToolResult:
        try:
            result = await manager.call_tool(
                await resolve_session(),
                namespace,
                name,
                arguments,
            )
            return ToolResult(
                content=result.content,
                structured_content=result.structured_content,
                is_error=result.is_error,
                meta=result.meta,
            )
        except McpUnavailable as error:
            raise ToolError(str(error)) from error
        except (httpx2.HTTPError, MCPError, TimeoutError) as error:
            raise ToolError("MCP upstream request failed") from error

    return mcp

octomate_mcp 

octomate_mcp(
    resolve_session: Callable[
        [], Awaitable[OctomateSession]
    ],
    thread_manager: ThreadManager,
    kick: Callable[
        [NativeGatewaySignal],
        Awaitable[Task[ReflexGraphResult] | None],
    ]
    | None = None,
    *,
    bearers: KnownBearers | None = None,
    manager: McpManager,
) -> FastMCP

Compose the gateway, history, and user-scoped MCP tools over one session resolver.

Source code in octomate/mcp/server.py 
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
def octomate_mcp(
    resolve_session: Callable[[], Awaitable[OctomateSession]],
    thread_manager: ThreadManager,
    kick: Callable[
        [NativeGatewaySignal], Awaitable[asyncio.Task[ReflexGraphResult] | None]
    ]
    | None = None,
    *,
    bearers: KnownBearers | None = None,
    manager: McpManager,
) -> FastMCP:
    """Compose the gateway, history, and user-scoped MCP tools over one session resolver."""
    session = Depends(resolve_session)
    mcp = FastMCP(
        name=OCTOMATE_SERVER_NAME,
        instructions=octomate_instructions(),
        auth=bearers,
    )
    gateway = FastMCP(GATEWAY_NAMESPACE)
    mount_gateway(gateway, session, thread_manager, kick)
    mcp.mount(gateway, namespace=GATEWAY_NAMESPACE)
    history = FastMCP(HISTORY_NAMESPACE)
    mount_history(history, session, thread_manager)
    mcp.mount(history, namespace=HISTORY_NAMESPACE)
    # Discovery, call, and linking helpers already carry their served names.
    mcp.mount(tentacles_mcp(resolve_session, manager=manager))
    return mcp

octomate.mcp.gateway

The gateway as MCP: the routing spells, served to every runtime that is not Inkling.

One family on the server octomate.mcp.server composes. Its tools are typed functions whose contracts are Inkling's own tool docstrings and whose schemas come from the same shapes, so no two runtimes read two different tools; all that differs is the session a call runs against, which each tool takes as a FastMCP dependency the mounting side supplied — Depends(...) of a fixed session for a server mounted in-process for one turn, of a per-request lookup for a server that answers over HTTP.

Refusals are spoken as the model reads them: the GatewayRefusal policy raises becomes a ToolError carrying the same sentence, as it becomes a ModelRetry for Inkling, so every runtime corrects from one wording.

TELEPORT_RECORDED module-attribute 

TELEPORT_RECORDED = "Teleporting — this turn ends here; you continue over there, with your context intact."

NATIVE_TELEPORT_STARTED module-attribute 

NATIVE_TELEPORT_STARTED = "Teleporting — this conversation continues at the destination, its history with it, in a session Octomate runs there. This session stays as it is."

CONVERSATION_HEADER module-attribute 

CONVERSATION_HEADER = 'X-Octomate-Conversation'

CLIENT_HEADER module-attribute 

CLIENT_HEADER = 'X-Octomate-Client'

spoken 

spoken(
    spell: Callable[SpellP, Awaitable[SpellT]],
) -> Callable[SpellP, Awaitable[SpellT]]

A spell with its refusal spoken as the model reads it: a ToolError carrying the gateway's sentence verbatim, as ModelRetry does for Inkling. A refusal is ordinary traffic, not a failure, and is logged as such — anything else that escapes a spell is a real error and keeps FastMCP's own handling.

Source code in octomate/mcp/gateway.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def spoken[**SpellP, SpellT](
    spell: Callable[SpellP, Awaitable[SpellT]],
) -> Callable[SpellP, Awaitable[SpellT]]:
    """A spell with its refusal spoken as the model reads it: a `ToolError` carrying
    the gateway's sentence verbatim, as `ModelRetry` does for Inkling. A refusal is
    ordinary traffic, not a failure, and is logged as such — anything else that
    escapes a spell is a real error and keeps FastMCP's own handling."""

    @wraps(spell)
    async def cast(*args: SpellP.args, **kwargs: SpellP.kwargs) -> SpellT:
        try:
            return await spell(*args, **kwargs)
        except GatewayRefusal as refusal:
            raise ToolError(str(refusal), log_level=logging.INFO) from refusal

    return cast

served_session 

served_session(
    octomate: Octomate,
) -> Callable[[], Awaitable[OctomateSession]]

The session a call served over HTTP runs against.

Every call speaks for the registered user its verified bearer names. A driven turn names its conversation with CONVERSATION_HEADER, written into its launch config by Octomate itself along with the kicker's own secret, and resolves to the session registered while that run is in flight — a bearer other than the kicker's is refused, so no user can drive another's turn. A native session names its runtime with CLIENT_HEADER, written once at install time, and gets native_session built for the bearer's user. Identity is asserted by config and credential either way, never chosen by the model, so a call carrying neither header is refused outright rather than guessed at.

Source code in octomate/mcp/gateway.py 
 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
def served_session(octomate: Octomate) -> Callable[[], Awaitable[OctomateSession]]:
    """The session a call served over HTTP runs against.

    Every call speaks for the registered user its verified bearer names. A driven
    turn names its conversation with `CONVERSATION_HEADER`, written into its
    launch config by Octomate itself along with the kicker's own secret, and
    resolves to the session registered while that run is in flight — a bearer
    other than the kicker's is refused, so no user can drive another's turn. A
    native session names its runtime with `CLIENT_HEADER`, written once at
    install time, and gets `native_session` built for the bearer's user.
    Identity is asserted by config and credential either way, never chosen by
    the model, so a call carrying neither header is refused outright rather than
    guessed at."""

    async def resolve() -> OctomateSession:
        headers = get_http_headers()
        access = get_access_token()
        principal = access.client_id if access is not None else None
        named = headers.get(CONVERSATION_HEADER.lower())
        if named is not None:
            try:
                conversation_id = uuid.UUID(named)
            except ValueError:
                raise ToolError(
                    f"{CONVERSATION_HEADER} is not a conversation id: {named!r}."
                ) from None
            session = octomate.gateway.get(conversation_id)
            if session is None:
                raise ToolError(
                    f"No turn of conversation {named} is at the gateway; a session "
                    "reaches it only while the run that opened it is in flight."
                )
            kicker = (
                await octomate.users.owner(session.user_profile)
                if session.user_profile is not None
                else None
            )
            if kicker is None or kicker.username != principal:
                raise ToolError(
                    f"Conversation {named}'s turn is not this bearer's to drive: "
                    "a driven session speaks with its kicker's own credential, "
                    "which its launch config carries."
                )
            return session
        client = headers.get(CLIENT_HEADER.lower())
        if client is None:
            raise ToolError(
                "This call names no identity: a driven turn names its conversation "
                f"with {CONVERSATION_HEADER}, a native session its runtime with "
                f"{CLIENT_HEADER}. Both are written by config, so a call carrying "
                "neither has nothing the gateway may answer for."
            )
        if client not in NATIVE_TENTACLE_IDS:
            raise ToolError(
                f"{CLIENT_HEADER} names no native runtime: {client!r}. An install "
                f"writes one of: {', '.join(sorted(NATIVE_TENTACLE_IDS))}."
            )
        if principal is None:
            agent = client.removesuffix("-native")
            raise ToolError(
                "A native session speaks for a registered user, and this call's "
                "bearer names none. Run `octomate configure --token` with "
                "their API token on their machine, then re-run "
                f"`octomate {agent} mcp "
                "install`."
            )
        return await native_session(octomate, client, principal)

    return resolve

native_session async 

native_session(
    octomate: Octomate, client: str, username: str
) -> OctomateSession

An ephemeral gateway for one native call, never registered.

username is the verified bearer's owner, so the session speaks for that person: anchored on a transient profile of theirs, its destinations are their own linked accounts. The call is still attributed to a runtime, never to a terminal session, so the session has no thread and no address — every destination is a crossing. A static MCP config puts the tools in every session once installed, and the bearer is the only control.

Source code in octomate/mcp/gateway.py 
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
async def native_session(
    octomate: Octomate, client: str, username: str
) -> OctomateSession:
    """An ephemeral gateway for one native call, never registered.

    `username` is the verified bearer's owner, so the session speaks for that
    person: anchored on a transient profile of theirs, its destinations are
    their own linked accounts. The call is still attributed to a runtime, never
    to a terminal session, so the session has no thread and no address — every
    destination is a crossing. A static MCP config puts the tools in every
    session once installed, and the bearer is the only control.
    """
    profile = await octomate.users.native_profile(client, username)
    if profile is None:
        raise RuntimeError(
            f"the bearer verified as {username!r} but the registry holds no such user"
        )
    return OctomateSession(
        channel_routes=octomate.gateway.available_routes(
            octomate.channels, octomate.agents
        ),
        current_agent_id=client,
        channels=octomate.channels,
        users=octomate.users,
        user_profile=profile,
        agents=octomate.agents,
        native=True,
        threads=octomate.threads,
        workspaces=octomate.workspaces,
        nerve=octomate.nerve,
    )

caller_thread 

caller_thread() -> str | None

The native session a call came from, where its runtime says so itself: Codex names its thread in every call's _meta.

Source code in octomate/mcp/gateway.py 
202
203
204
205
206
207
208
def caller_thread() -> str | None:
    """The native session a call came from, where its runtime says so itself:
    Codex names its thread in every call's `_meta`."""
    request = get_context().request_context
    meta = request.meta if request is not None else None
    thread_id = (meta or {}).get("threadId")
    return thread_id if isinstance(thread_id, str) and thread_id else None

mount_gateway 

mount_gateway(
    mcp: FastMCP,
    octomate_session: OctomateSession,
    thread_manager: ThreadManager,
    kick: Callable[
        [NativeGatewaySignal],
        Awaitable[Task[ReflexGraphResult] | None],
    ]
    | None = None,
) -> None

Register the gateway's spells on mcp.

octomate_session is the FastMCP dependency each call resolves its session through — Depends(...) of a fixed session for a server mounted in-process for one turn, of a per-request lookup for a server that answers over HTTP. thread_manager is the ledger a delivering spell writes through. kick is how a native session's teleport or scheme becomes its own turn at once, so only the served mount — the one place a native session can arrive — needs one.

Source code in octomate/mcp/gateway.py 
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
def mount_gateway(
    mcp: FastMCP,
    octomate_session: OctomateSession,
    thread_manager: ThreadManager,
    kick: Callable[
        [NativeGatewaySignal], Awaitable[asyncio.Task[ReflexGraphResult] | None]
    ]
    | None = None,
) -> None:
    """Register the gateway's spells on `mcp`.

    `octomate_session` is the FastMCP dependency each call resolves its session
    through — `Depends(...)` of a fixed session for a server mounted in-process for
    one turn, of a per-request lookup for a server that answers over HTTP.
    `thread_manager` is the ledger a delivering spell writes through. `kick` is how
    a native session's teleport or scheme becomes its own turn at once, so only the
    served mount — the one place a native session can arrive — needs one.
    """

    @mcp.tool(
        name=GatewayTool.INSPECT,
        description=capability_contract(GatewayCapability.inspect),
        annotations={"readOnlyHint": True},
    )
    @spoken
    async def inspect(
        reveal: InspectFacet,
        channel: str | None = None,
        inside: str | None = None,
        session: OctomateSession = octomate_session,
    ) -> str:
        if reveal == "destinations":
            addresses = await session.inspect("destinations", channel, inside)
            return (
                TypeAdapter(list[ChannelAddress]).dump_json(addresses).decode()
                if addresses
                else "- (none)"
            )
        # Lines, never the list: FastMCP renders an empty list as no content at all.
        return (
            "\n".join(
                str(one) for one in await session.inspect(reveal, channel, inside)
            )
            or "- (none)"
        )

    @mcp.tool(
        name=GatewayTool.SUMMON,
        description=capability_contract(GatewayCapability.summon),
    )
    @spoken
    async def summon(
        agent_id: str,
        model: str,
        hint: str,
        reason: str,
        brief: Annotated[str, Field(max_length=8_000)],
        effort: str | None = None,
        session: OctomateSession = octomate_session,
    ) -> str:
        return await session.summon(
            agent_id=agent_id,
            model=model,
            hint=hint,
            reason=reason,
            brief=brief,
            effort=effort,
        )

    @mcp.tool(
        name=GatewayTool.TELEPORT,
        description=capability_contract(GatewayCapability.teleport),
    )
    @spoken
    async def teleport(
        hint: str,
        destination: ChannelAddress | None = None,
        project: str | None = None,
        ref: str | None = None,
        new_thread: bool = True,
        resume: bool = False,
        session_id: Annotated[
            str | None,
            Field(
                description="Filled in by Octomate's own hook for a session you "
                "were not started by Octomate in; leave it out."
            ),
        ] = None,
        session: OctomateSession = octomate_session,
    ) -> str:
        native_session_id = (session_id or caller_thread()) if session.native else None
        if native_session_id is not None:
            await session.attach_native_thread(native_session_id)
        await session.teleport(
            hint=hint,
            destination=destination,
            new_thread=new_thread,
            project=project,
            ref=ref,
            resume=resume,
        )
        if not session.native:
            return TELEPORT_RECORDED
        if kick is None:
            raise RuntimeError(
                "a native session reached a gateway mounted without a kick"
            )
        await kick(session.native_handoff())
        return NATIVE_TELEPORT_STARTED

    @mcp.tool(
        name=GatewayTool.SCHEME,
        description=capability_contract(GatewayCapability.scheme),
    )
    @spoken
    async def scheme(
        hint: str,
        brief: Annotated[str, Field(max_length=8_000)],
        destination: SchemeTarget = DIRECT_TARGET,
        session: OctomateSession = octomate_session,
    ) -> str:
        sentence = await session.scheme(hint=hint, brief=brief, destination=destination)
        if session.native:
            if kick is None:
                raise RuntimeError(
                    "a native session reached a gateway mounted without a kick"
                )
            await kick(session.native_handoff())
        return sentence

    @mcp.tool(
        name=GatewayTool.SEND, description=capability_contract(GatewayCapability.send)
    )
    @spoken
    async def send(
        segments: list[MessageSegment],
        destination: SendTarget = HERE_TARGET,
        session: OctomateSession = octomate_session,
    ) -> str:
        # Published right here: a gateway `send` has no run stream to ride, so what
        # React does for Inkling's sends happens in the call instead — the same
        # resolve, the same DM open, the same ledger row, then the channel delivers.
        address = await session.resolve_send(destination)
        target = address if address is not None else session.conversation_address
        if target is None:
            raise GatewayRefusal(
                "This session has no conversation of its own to land a send on — "
                'use a destination with kind="dm" and an explicit connected channel ID.'
            )
        notice = "sent"
        private = False
        if address is not None:
            channel = session.channels[target.channel_tentacle_id]
            dm = await channel.open_dm(target.user_id)
            if dm is not None:
                target = dm
                private = True
            else:
                fallback = session.conversation_address
                if fallback is None:
                    raise GatewayRefusal(
                        f"{channel.name} could not open a direct message with "
                        f"{target.user_id!r}; nothing was delivered."
                    )
                # Mirror React's own fallback: land it where the run lives, and say so.
                target = fallback
                notice = (
                    f"could not open their direct messages on {channel.name}; "
                    "delivered to this conversation instead"
                )
        channel = session.channels[target.channel_tentacle_id]
        message = await thread_manager.record_outbound(
            target,
            agent_tentacle_id=session.current_agent_id,
            segments=segments,
            sender=channel.self_profile,
        )
        sent = MessageSentEvent(
            segments=segments, destination=target, thread_message_id=message.id
        )
        await session.publish(target, sent, private=private)
        return notice

    @mcp.tool(
        name=GatewayTool.DISMISS,
        description=capability_contract(GatewayCapability.dismiss),
    )
    @spoken
    async def dismiss(session: OctomateSession = octomate_session) -> str:
        return await session.dismiss()

octomate.mcp.history

The history tools as MCP: the thread ledger of the person a session speaks for, served beside the spells.

A family of its own on the one server, mounted under the history namespace. Every return is text and bounded — a page of at most HISTORY_PAGE_LIMIT messages, each clipped to HISTORY_LINE_CHARS — because these runtimes have none of the spill bands that catch an oversized return for Inkling.

HISTORY_LINE_CHARS module-attribute 

HISTORY_LINE_CHARS = 400

HISTORY_PAGE_LIMIT module-attribute 

HISTORY_PAGE_LIMIT = 50

HISTORY_TOOL_NAMES module-attribute 

HISTORY_TOOL_NAMES: dict[str, str] = {
    "search_thread_history": "search",
    "read_thread_history_before": "read_before",
    "read_thread_history_after": "read_after",
}

mount_history 

mount_history(
    mcp: FastMCP,
    octomate_session: OctomateSession,
    thread_manager: ThreadManager,
) -> None

Register the history tools on mcp, every call reading as the person the session octomate_session resolves to speaks for — a driven turn's user, or the registered person a native session's bearer named.

Source code in octomate/mcp/history.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
def mount_history(
    mcp: FastMCP, octomate_session: OctomateSession, thread_manager: ThreadManager
) -> None:
    """Register the history tools on `mcp`, every call reading as the person the
    session `octomate_session` resolves to speaks for — a driven turn's user, or the
    registered person a native session's bearer named."""

    def reader(session: OctomateSession) -> UserProfile:
        if session.user_profile is None:
            raise ToolError(
                "This session speaks for nobody, and history is a person's: there "
                "is no one whose threads to read."
            )
        return session.user_profile

    async def anchor(session: OctomateSession, handle: str) -> ThreadMessage:
        try:
            return await thread_manager.chat_message(reader(session), handle)
        except ValueError as refusal:
            raise ToolError(str(refusal), log_level=logging.INFO) from refusal

    def page(limit: int) -> int:
        if limit > HISTORY_PAGE_LIMIT:
            raise ToolError(
                f"limit {limit} is over the page of {HISTORY_PAGE_LIMIT}; page in "
                "smaller steps."
            )
        return limit

    @mcp.tool(
        name=HISTORY_TOOL_NAMES["search_thread_history"],
        description=capability_contract(HistoryCapability.search_thread_history),
        annotations={"readOnlyHint": True},
    )
    async def search_thread_history(
        query: str,
        actor_kind: ChannelActorKind | None = None,
        limit: int = 10,
        session: OctomateSession = octomate_session,
    ) -> str:
        try:
            rows = await thread_manager.search_chat_messages(
                reader(session), query, actor_kind=actor_kind, limit=page(limit)
            )
        except ValueError as refusal:
            raise ToolError(str(refusal), log_level=logging.INFO) from refusal
        return (
            "\n".join(
                line
                if len(line) <= HISTORY_LINE_CHARS
                else line[:HISTORY_LINE_CHARS] + "…"
                for line in map(str, rows)
            )
            or "(no messages)"
        )

    @mcp.tool(
        name=HISTORY_TOOL_NAMES["read_thread_history_before"],
        description=capability_contract(HistoryCapability.read_thread_history_before),
        annotations={"readOnlyHint": True},
    )
    async def read_thread_history_before(
        message_id: str, limit: int = 10, session: OctomateSession = octomate_session
    ) -> str:
        found = await anchor(session, message_id)
        rows = await thread_manager.chat_messages_before(
            found.thread_id, found.id, limit=page(limit)
        )
        return (
            "\n".join(
                line
                if len(line) <= HISTORY_LINE_CHARS
                else line[:HISTORY_LINE_CHARS] + "…"
                for line in map(str, rows)
            )
            or "(no messages)"
        )

    @mcp.tool(
        name=HISTORY_TOOL_NAMES["read_thread_history_after"],
        description=capability_contract(HistoryCapability.read_thread_history_after),
        annotations={"readOnlyHint": True},
    )
    async def read_thread_history_after(
        message_id: str, limit: int = 10, session: OctomateSession = octomate_session
    ) -> str:
        found = await anchor(session, message_id)
        rows = await thread_manager.chat_messages_after(
            found.thread_id, found.id, limit=page(limit)
        )
        return (
            "\n".join(
                line
                if len(line) <= HISTORY_LINE_CHARS
                else line[:HISTORY_LINE_CHARS] + "…"
                for line in map(str, rows)
            )
            or "(no messages)"
        )

octomate.mcp.oauth

Link the current channel profile and authorize the user's installed MCPs.

Authorization links and device codes go to the user's direct messages. The model receives status only; device flows finish through confirm.

OAUTH_NAMESPACE module-attribute 

OAUTH_NAMESPACE = 'oauth'

CONNECT_TOOL module-attribute 

CONNECT_TOOL = f'{OAUTH_NAMESPACE}_connect'

CONFIRM_TOOL module-attribute 

CONFIRM_TOOL = f'{OAUTH_NAMESPACE}_confirm'

LINK_PROFILE_TOOL = f'{OAUTH_NAMESPACE}_link_profile'

ProviderId module-attribute 

ProviderId = Annotated[
    str,
    Field(
        description="The user's installed MCP namespace."
    ),
]

mount_oauth 

mount_oauth(
    mcp: FastMCP,
    octomate_session: OctomateSession,
    manager: McpManager,
) -> None

Link profiles and authorize MCPs for the person driving the current turn.

Source code in octomate/mcp/oauth.py 
 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
208
209
210
211
212
213
214
215
def mount_oauth(
    mcp: FastMCP,
    octomate_session: OctomateSession,
    manager: McpManager,
) -> None:
    """Link profiles and authorize MCPs for the person driving the current turn."""

    async def identity(session: OctomateSession) -> tuple[User, UserProfile]:
        if session.user_profile is None:
            raise ToolError(
                "A link authorizes the person who drove this turn, and nobody "
                "registered did."
            )
        user = await manager.users.owner(session.user_profile)
        if user is None:
            raise ToolError("OAuth connections require a registered user")
        return user, session.user_profile

    async def personal(provider: str, user: User) -> OAuthMcp:
        try:
            return await manager.authorizable(user_id=user.id, namespace=provider)
        except McpUnavailable as error:
            raise ToolError("This OAuth MCP is unavailable") from error

    @mcp.tool(
        name="connect",
        description=(
            "Send this user a link that authorizes their own account with a "
            "provider identified by its installed MCP namespace. The link, and a code where the provider asks "
            "for one, go to their direct messages, never to the conversation, and "
            "are not returned here."
        ),
    )
    async def connect(
        provider: ProviderId,
        session: OctomateSession = octomate_session,
        *,
        flow: OAuthFlowKind | None = None,
    ) -> str:
        user, profile = await identity(session)
        address = session.conversation_address
        channel = (
            session.channels.get(address.channel_tentacle_id)
            if address is not None
            else None
        )
        if address is None or channel is None:
            raise ToolError(
                "The link goes to the person's direct messages on the channel "
                "this turn is on, and this call has no turn on a channel."
            )
        instance = await personal(provider, user)
        try:
            authorization = await manager.connect(
                user, instance.id, profile=profile, flow=flow
            )
        except McpUnavailable as error:
            raise ToolError(str(error)) from error
        label = instance.name
        # The authorization goes to the channel as an event of its own, for the
        # channel to present — never through this return value, which the model
        # reads and could repeat into a reply.
        if isinstance(authorization, AuthorizationLink):
            event = OAuthAuthorizationEvent(
                connector_id=provider,
                label=label,
                authorization_uri=str(authorization.authorization_uri),
            )
            sent = "The authorization link is on its way"
        else:
            event = OAuthDeviceAuthorizationEvent(
                connector_id=provider,
                label=label,
                authorization_uri=str(
                    authorization.verification_uri_complete
                    or authorization.verification_uri
                ),
                user_code=authorization.user_code.get_secret_value(),
            )
            sent = "The authorization link and code are on their way"
        # Settled here, so the link is heard only where the channel will show it.
        private = await channel.feelers.oauth.deliver_to(address)
        await session.publish(private, event, private=True)
        return f"{sent} to this user's direct messages."

    @mcp.tool(
        name="confirm",
        description="Check authorization for a user's installed MCP namespace, polling when required.",
    )
    async def confirm(
        provider: ProviderId, session: OctomateSession = octomate_session
    ) -> str:
        user, profile = await identity(session)
        instance = await personal(provider, user)
        status = await manager.confirm(user, instance.id, profile=profile)
        match status:
            case McpAuthorizationStatus(status="active"):
                return f"{provider} is connected and available through namespace discovery."
            case McpDeviceAuthorizationPending(retry_after_seconds=delay):
                return f"{provider} is waiting for approval; confirm again in {delay} seconds."
            case McpBrowserAuthorizationPending():
                return f"{provider} is waiting for approval in the browser. Confirm after approval."
            case _:
                return f"{provider} needs authorization. Call `{CONNECT_TOOL}` with `{provider}`."

    @mcp.tool(
        name="link_profile",
        description=(
            "Request authorization from the Octomate host to link the channel profile "
            "driving this turn to the signed-in user's account. The host authorization "
            "card is sent privately through the channel. Use this when the "
            "user asks to link their current channel profile. The profile and "
            "destination come from the current turn, never tool arguments; the "
            "link goes to that user's direct messages and is not returned here. "
            "Trunkline already uses the signed-in Octomate identity and cannot link profiles."
        ),
    )
    async def link_profile(session: OctomateSession = octomate_session) -> str:
        # Trunkline's routers import Octomate, which mounts this MCP module.
        from octomate.tentacles.trunkline.base import TrunklineTentacle

        profile = session.user_profile
        address = session.conversation_address
        channel = (
            session.channels.get(address.channel_tentacle_id)
            if address is not None
            else None
        )
        if isinstance(channel, TrunklineTentacle):
            raise ToolError(
                "Trunkline already uses your signed-in Octomate identity; "
                "profile linking is not needed."
            )
        if profile is None or address is None or channel is None:
            raise ToolError(
                "Linking a profile requires a turn from a user on a channel."
            )
        if (
            profile.channel_tentacle_id != address.channel_tentacle_id
            or profile.channel_user_id != address.user_id
        ):
            raise ToolError(
                "The current channel profile does not match the user who drove this turn."
            )
        try:
            private_address = await channel.feelers.oauth.deliver_to(address)
        except RuntimeError as error:
            raise ToolError(
                "This channel has no private surface for a profile link."
            ) from error
        try:
            authorization = await manager.users.start_link_profile(profile)
        except (
            InvalidLinkProfile,
            ProfileAlreadyLinked,
            LinkProfileUnavailable,
        ) as error:
            raise ToolError(str(error)) from error
        message_id = await channel.feelers.oauth.present(
            private_address,
            LinkProfileAuthorizationEvent(
                host=channel.octomate.title,
                authorization=authorization,
            ),
        )
        if message_id is None:
            raise ToolError("The channel could not deliver the private profile link.")
        return "The profile link was sent privately to this user."

octomate.mcp.base

What every Octomate MCP server shares: the bearer verifier for scoped API tokens, and the docstring projection that keeps a spell's MCP tool and its Inkling tool identical.

KnownBearers

Bases: TokenVerifier

Verify scoped API tokens through the application's auth manager.

Source code in octomate/mcp/base.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
class KnownBearers(TokenVerifier):
    """Verify scoped API tokens through the application's auth manager."""

    def __init__(self, auth: AuthManager | None = None) -> None:
        super().__init__()
        self.auth: AuthManager | None = auth

    async def owner(self, token: str, *, scope: ApiKeyScope) -> str | None:
        if self.auth is None:
            return None
        key = await self.auth.authenticate_api_key(SecretStr(token), scope=scope)
        if key is None:
            return None
        async with async_session() as session:
            user = await session.get(User, key.user_id)
        return user.username if user is not None else None

    async def verify_token(self, token: str) -> AccessToken | None:
        principal = await self.owner(token, scope="mcp")
        if principal is None:
            return None
        return AccessToken(token=token, client_id=principal, scopes=["mcp"])

auth instance-attribute 

auth: AuthManager | None = auth

owner async 

owner(token: str, *, scope: ApiKeyScope) -> str | None
Source code in octomate/mcp/base.py 
26
27
28
29
30
31
32
33
34
async def owner(self, token: str, *, scope: ApiKeyScope) -> str | None:
    if self.auth is None:
        return None
    key = await self.auth.authenticate_api_key(SecretStr(token), scope=scope)
    if key is None:
        return None
    async with async_session() as session:
        user = await session.get(User, key.user_id)
    return user.username if user is not None else None

verify_token async 

verify_token(token: str) -> AccessToken | None
Source code in octomate/mcp/base.py 
36
37
38
39
40
async def verify_token(self, token: str) -> AccessToken | None:
    principal = await self.owner(token, scope="mcp")
    if principal is None:
        return None
    return AccessToken(token=token, client_id=principal, scopes=["mcp"])

capability_contract 

capability_contract(
    spell: Callable[..., Awaitable[object]],
) -> str

The docstring Inkling's toolset compiles, verbatim — a copy here would give two models two different tools and drift silently.

Source code in octomate/mcp/base.py 
43
44
45
46
47
48
49
def capability_contract(spell: Callable[..., Awaitable[object]]) -> str:
    """The docstring Inkling's toolset compiles, verbatim — a copy here would give
    two models two different tools and drift silently."""
    doc = spell.__doc__
    if doc is None:
        raise RuntimeError(f"{spell.__qualname__} has no docstring to project")
    return cleandoc(doc)

octomate.mcp.transport

An httpx transport for MCP endpoints that connects only to public addresses and refuses redirects.

IP_ADDRESSES module-attribute 

IP_ADDRESSES = TypeAdapter(list[IPvAnyAddress])

PublicMcpNetwork

Bases: AnyIOBackend

Resolve and pin each connection to a public address, including reconnects.

Source code in octomate/mcp/transport.py 
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
class PublicMcpNetwork(AnyIOBackend):
    """Resolve and pin each connection to a public address, including reconnects."""

    async def connect_tcp(
        self,
        host: str,
        port: int,
        # httpcore's backend contract requires this parameter.
        timeout: float | None = None,  # noqa: ASYNC109
        local_address: str | None = None,
        socket_options: Iterable[SOCKET_OPTION] | None = None,
    ) -> AsyncNetworkStream:
        async with asyncio.timeout(timeout):
            addresses = await asyncio.get_running_loop().getaddrinfo(
                host=host, port=port, type=socket.SOCK_STREAM
            )
            ips = IP_ADDRESSES.validate_python([address[4][0] for address in addresses])
            if not ips or any(not ip.is_global or ip.is_multicast for ip in ips):
                raise ConnectError(
                    "MCP endpoints must resolve only to public addresses"
                )
            # Pass the checked numeric address to the socket layer; TLS still uses
            # the original hostname held by httpcore's connection.
            return await super().connect_tcp(
                host=str(addresses[0][4][0]),
                port=port,
                timeout=timeout,
                local_address=local_address,
                socket_options=socket_options,
            )

connect_tcp async 

connect_tcp(
    host: str,
    port: int,
    timeout: float | None = None,
    local_address: str | None = None,
    socket_options: Iterable[SOCKET_OPTION] | None = None,
) -> AsyncNetworkStream
Source code in octomate/mcp/transport.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
47
48
49
async def connect_tcp(
    self,
    host: str,
    port: int,
    # httpcore's backend contract requires this parameter.
    timeout: float | None = None,  # noqa: ASYNC109
    local_address: str | None = None,
    socket_options: Iterable[SOCKET_OPTION] | None = None,
) -> AsyncNetworkStream:
    async with asyncio.timeout(timeout):
        addresses = await asyncio.get_running_loop().getaddrinfo(
            host=host, port=port, type=socket.SOCK_STREAM
        )
        ips = IP_ADDRESSES.validate_python([address[4][0] for address in addresses])
        if not ips or any(not ip.is_global or ip.is_multicast for ip in ips):
            raise ConnectError(
                "MCP endpoints must resolve only to public addresses"
            )
        # Pass the checked numeric address to the socket layer; TLS still uses
        # the original hostname held by httpcore's connection.
        return await super().connect_tcp(
            host=str(addresses[0][4][0]),
            port=port,
            timeout=timeout,
            local_address=local_address,
            socket_options=socket_options,
        )

PublicMcpTransport

Bases: AsyncHTTPTransport

Source code in octomate/mcp/transport.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
class PublicMcpTransport(httpx2.AsyncHTTPTransport):
    def __init__(self) -> None:
        # HTTPX exposes no network-backend argument. Keep its request/response
        # adapter and supply the httpcore pool at this one dependency boundary.
        self._pool = AsyncConnectionPool(
            ssl_context=ssl.create_default_context(),
            network_backend=PublicMcpNetwork(),
        )

    async def handle_async_request(self, request: httpx2.Request) -> httpx2.Response:
        response = await super().handle_async_request(request)
        if response.has_redirect_location:
            await response.aclose()
            raise httpx2.HTTPStatusError(
                "Install the final MCP endpoint URL; redirects are not allowed",
                request=request,
                response=response,
            )
        return response

handle_async_request async 

handle_async_request(request: Request) -> httpx2.Response
Source code in octomate/mcp/transport.py 
61
62
63
64
65
66
67
68
69
70
async def handle_async_request(self, request: httpx2.Request) -> httpx2.Response:
    response = await super().handle_async_request(request)
    if response.has_redirect_location:
        await response.aclose()
        raise httpx2.HTTPStatusError(
            "Install the final MCP endpoint URL; redirects are not allowed",
            request=request,
            response=response,
        )
    return response

mcp_http_client 

mcp_http_client(
    headers: dict[str, str] | None = None,
    timeout: Timeout | None = None,
    auth: Auth | None = None,
    follow_redirects: bool = False,
) -> httpx2.AsyncClient
Source code in octomate/mcp/transport.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def mcp_http_client(
    headers: dict[str, str] | None = None,
    timeout: httpx2.Timeout | None = None,
    auth: httpx2.Auth | None = None,
    follow_redirects: bool = False,
) -> httpx2.AsyncClient:
    return httpx2.AsyncClient(
        headers=headers,
        timeout=timeout or httpx2.Timeout(30),
        auth=auth,
        transport=PublicMcpTransport(),
        trust_env=False,
        # A redirect must never carry an instance's credential to another
        # endpoint. Users install the final MCP URL explicitly.
        follow_redirects=False,
    )

octomate.mcp.routes

The HTTP API for a user's MCP installations: listing, installing, authorizing, enabling and removing them.

logger module-attribute 

logger = logging.getLogger(__name__)

mcp_router module-attribute 

mcp_router = APIRouter(
    prefix="/api/mcp",
    tags=["mcp"],
    dependencies=[Depends(browser_request)],
)

available_tentacles async 

available_tentacles(
    user: Annotated[User, Depends(current_user)],
    manager: Annotated[McpManager, Depends(mcp_manager)],
) -> list[McpTentacleInfo]
Source code in octomate/mcp/routes.py 
36
37
38
39
40
41
@mcp_router.get("/tentacles", response_model=list[McpTentacleInfo])
async def available_tentacles(
    user: Annotated[User, Depends(current_user)],
    manager: Annotated[McpManager, Depends(mcp_manager)],
) -> list[McpTentacleInfo]:
    return manager.available()

list_instances async 

list_instances(
    user: Annotated[User, Depends(current_user)],
    manager: Annotated[McpManager, Depends(mcp_manager)],
) -> list[Mcp]
Source code in octomate/mcp/routes.py 
44
45
46
47
48
49
@mcp_router.get("", response_model=list[McpVariant])
async def list_instances(
    user: Annotated[User, Depends(current_user)],
    manager: Annotated[McpManager, Depends(mcp_manager)],
) -> list[Mcp]:
    return await manager.list(user.id)

install async 

install(
    body: McpInstallRequest,
    user: Annotated[User, Depends(current_user)],
    manager: Annotated[McpManager, Depends(mcp_manager)],
) -> Mcp
Source code in octomate/mcp/routes.py 
52
53
54
55
56
57
58
59
60
61
62
63
@mcp_router.post("", status_code=201, response_model=McpVariant)
async def install(
    body: McpInstallRequest,
    user: Annotated[User, Depends(current_user)],
    manager: Annotated[McpManager, Depends(mcp_manager)],
) -> Mcp:
    try:
        return await manager.install(user.id, body)
    except McpUnavailable as error:
        raise HTTPException(status_code=404, detail=str(error)) from error
    except ValueError as error:
        raise HTTPException(status_code=400, detail=str(error)) from error

pending_authorization async 

pending_authorization(
    mcp_id: UUID7,
    user: Annotated[User, Depends(current_user)],
    manager: Annotated[McpManager, Depends(mcp_manager)],
) -> JSONResponse
Source code in octomate/mcp/routes.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
@mcp_router.get("/{mcp_id}/authorization", response_model=OAuthStartResult | None)
async def pending_authorization(
    mcp_id: UUID7,
    user: Annotated[User, Depends(current_user)],
    manager: Annotated[McpManager, Depends(mcp_manager)],
) -> JSONResponse:
    try:
        authorization = await manager.pending_authorization(user, mcp_id)
        content = (
            authorization.model_dump(mode="json") if authorization is not None else None
        )
        if isinstance(authorization, DeviceAuthorization):
            assert content is not None
            content["user_code"] = authorization.user_code.get_secret_value()
        return JSONResponse(content=content, headers={"Cache-Control": "no-store"})
    except McpUnavailable as error:
        raise HTTPException(status_code=404, detail=str(error)) from error
    except ValueError as error:
        raise HTTPException(
            status_code=400, detail="Pending MCP authorization could not be restored"
        ) from error

cancel_authorization async 

cancel_authorization(
    mcp_id: UUID7,
    user: Annotated[User, Depends(current_user)],
    manager: Annotated[McpManager, Depends(mcp_manager)],
) -> None
Source code in octomate/mcp/routes.py 
89
90
91
92
93
94
95
96
97
98
@mcp_router.delete("/{mcp_id}/authorization", status_code=204)
async def cancel_authorization(
    mcp_id: UUID7,
    user: Annotated[User, Depends(current_user)],
    manager: Annotated[McpManager, Depends(mcp_manager)],
) -> None:
    try:
        await manager.cancel_authorization(user, mcp_id)
    except McpUnavailable as error:
        raise HTTPException(status_code=404, detail=str(error)) from error

connect async 

connect(
    mcp_id: UUID7,
    user: Annotated[User, Depends(current_user)],
    manager: Annotated[McpManager, Depends(mcp_manager)],
    flow: OAuthFlowKind | None = None,
) -> JSONResponse
Source code in octomate/mcp/routes.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
@mcp_router.post("/{mcp_id}/connect", response_model=OAuthStartResult)
async def connect(
    mcp_id: UUID7,
    user: Annotated[User, Depends(current_user)],
    manager: Annotated[McpManager, Depends(mcp_manager)],
    flow: OAuthFlowKind | None = None,
) -> JSONResponse:
    try:
        authorization = await manager.connect(user, mcp_id, flow=flow)
    except McpUnavailable as error:
        raise HTTPException(status_code=404, detail=str(error)) from error
    except (ValueError, httpx2.HTTPError, OAuthFlowError) as error:
        logger.warning("Failed to authorize MCP %s (%s)", mcp_id, type(error).__name__)
        raise HTTPException(
            status_code=400, detail="MCP authorization could not be started"
        ) from error
    content = authorization.model_dump(mode="json")
    if isinstance(authorization, DeviceAuthorization):
        content["user_code"] = authorization.user_code.get_secret_value()
    return JSONResponse(content=content, headers={"Cache-Control": "no-store"})

confirm async 

confirm(
    mcp_id: UUID7,
    user: Annotated[User, Depends(current_user)],
    manager: Annotated[McpManager, Depends(mcp_manager)],
) -> McpAuthorizationResult
Source code in octomate/mcp/routes.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
@mcp_router.post("/{mcp_id}/confirm", response_model=McpAuthorizationResult)
async def confirm(
    mcp_id: UUID7,
    user: Annotated[User, Depends(current_user)],
    manager: Annotated[McpManager, Depends(mcp_manager)],
) -> McpAuthorizationResult:
    try:
        return await manager.confirm(user, mcp_id)
    except McpUnavailable as error:
        raise HTTPException(status_code=404, detail=str(error)) from error
    except (ValueError, httpx2.HTTPError, OAuthFlowError) as error:
        logger.warning("Failed to confirm MCP %s (%s)", mcp_id, type(error).__name__)
        raise HTTPException(
            status_code=400, detail="MCP authorization could not be confirmed"
        ) from error

enable async 

enable(
    mcp_id: UUID7,
    user: Annotated[User, Depends(current_user)],
    manager: Annotated[McpManager, Depends(mcp_manager)],
) -> Mcp
Source code in octomate/mcp/routes.py 
140
141
142
143
144
145
146
147
148
149
@mcp_router.post("/{mcp_id}/enable", response_model=McpVariant)
async def enable(
    mcp_id: UUID7,
    user: Annotated[User, Depends(current_user)],
    manager: Annotated[McpManager, Depends(mcp_manager)],
) -> Mcp:
    try:
        return await manager.enable(user_id=user.id, mcp_id=mcp_id)
    except McpUnavailable as error:
        raise HTTPException(status_code=404, detail=str(error)) from error

disable async 

disable(
    mcp_id: UUID7,
    user: Annotated[User, Depends(current_user)],
    manager: Annotated[McpManager, Depends(mcp_manager)],
) -> Mcp
Source code in octomate/mcp/routes.py 
152
153
154
155
156
157
158
159
160
161
@mcp_router.post("/{mcp_id}/disable", response_model=McpVariant)
async def disable(
    mcp_id: UUID7,
    user: Annotated[User, Depends(current_user)],
    manager: Annotated[McpManager, Depends(mcp_manager)],
) -> Mcp:
    try:
        return await manager.disable(user_id=user.id, mcp_id=mcp_id)
    except McpUnavailable as error:
        raise HTTPException(status_code=404, detail=str(error)) from error

uninstall async 

uninstall(
    mcp_id: UUID7,
    user: Annotated[User, Depends(current_user)],
    manager: Annotated[McpManager, Depends(mcp_manager)],
) -> None
Source code in octomate/mcp/routes.py 
164
165
166
167
168
169
170
171
172
173
@mcp_router.delete("/{mcp_id}", status_code=204)
async def uninstall(
    mcp_id: UUID7,
    user: Annotated[User, Depends(current_user)],
    manager: Annotated[McpManager, Depends(mcp_manager)],
) -> None:
    try:
        await manager.uninstall(user.id, mcp_id)
    except McpUnavailable as error:
        raise HTTPException(status_code=404, detail=str(error)) from error