wip: support stdio mcp

Signed-off-by: Jan Pokorný <JenomPokorny@gmail.com>

Author: JanPokorny

apps/agentstack-server/Dockerfile

@@ -1,7 +1,9 @@
-FROM python:3.13-alpine3.22 AS builder
+FROM python:3.13-alpine3.22
 ENV UV_COMPILE_BYTECODE=1 \
     HOME="/tmp" \
     AGENT_REGISTRY__LOCATIONS__FILE="file:///app/registry.yaml"
+RUN apk add --no-cache kubectl nodejs npm \
+    && npm install -g supergateway
 RUN --mount=type=cache,target=/tmp/.cache/uv \
     --mount=type=bind,source=dist/requirements.txt,target=/requirements.txt \
     --mount=type=bind,from=ghcr.io/astral-sh/uv:0.9.5,source=/uv,target=/bin/uv \

apps/agentstack-server/src/agentstack_server/api/routes/connectors.py

@@ -51,6 +51,7 @@ async def create_connector(
             client_secret=request.client_secret,
             metadata=request.metadata,
             match_preset=request.match_preset,
+            bearer_token=request.bearer_token,
         )
     )
 

apps/agentstack-server/src/agentstack_server/api/schema/connector.py

@@ -19,6 +19,7 @@ class ConnectorCreateRequest(BaseModel):
     metadata: Metadata | None = None
 
     match_preset: bool = True
+    bearer_token: str | None = None
 
 
 class AuthorizationCodeRequest(BaseModel):

apps/agentstack-server/src/agentstack_server/configuration.py

@@ -219,15 +219,31 @@ class ManagedProviderConfiguration(BaseModel):
     )
 
 
+class ConnectorStdioPreset(BaseModel):
+    image: str
+    command: list[str] | None = None
+    args: list[str] | None = None
+    env: dict[str, str] = Field(default_factory=dict)
+
+
 class ConnectorPreset(BaseModel):
     url: AnyUrl
     client_id: str | None = None
     client_secret: str | None = None
     metadata: dict[str, str] | None = None
+    bearer_token: str | None = None
+    stdio: ConnectorStdioPreset | None = None
+
+
+class ConnectorRuntimeConfiguration(BaseModel):
+    kubeconfig: Path | None = None
+    namespace: str | None = None
+    startup_timeout_seconds: int = 60
 
 
 class ConnectorConfiguration(BaseModel):
     presets: list[ConnectorPreset] = Field(default_factory=list)
+    runtime: ConnectorRuntimeConfiguration = Field(default_factory=ConnectorRuntimeConfiguration)
 
 
 class DoclingExtractionConfiguration(BaseModel):

apps/agentstack-server/src/agentstack_server/service_layer/services/connector.py

@@ -3,11 +3,14 @@
 
 from __future__ import annotations
 
+import asyncio
 import html
 import logging
-from contextlib import AsyncExitStack
+import shlex
+import socket
+from contextlib import AsyncExitStack, asynccontextmanager
 from datetime import timedelta
-from secrets import token_urlsafe
+from secrets import token_hex, token_urlsafe
 from urllib.parse import parse_qs, urlencode, urlparse, urlunparse
 from uuid import UUID
 
@@ -54,13 +57,17 @@ async def create_connector(
         client_secret: str | None,
         metadata: Metadata | None,
         match_preset: bool = True,
+        bearer_token: str | None = None,
     ) -> Connector:
         if client_secret and not client_id:
             raise PlatformError(
                 "client_id must be present when client_secret is specified", status_code=status.HTTP_400_BAD_REQUEST
             )
 
         preset = self._find_preset(url=url) if match_preset else None
+        if not preset and url.scheme not in {"http", "https"}:
+            raise PlatformError("Unknown connector preset", status_code=status.HTTP_400_BAD_REQUEST)
+
         if preset:
             if not client_id:
                 client_id = preset.client_id
@@ -72,6 +79,7 @@ async def create_connector(
             created_by=user.id,
             auth=Authorization(client_id=client_id, client_secret=client_secret) if client_id else None,
             metadata=metadata,
+            bearer_token=bearer_token,
         )
         async with self._uow() as uow:
             await uow.connectors.create(connector=connector)
@@ -368,10 +376,19 @@ def client_factory(headers=None, timeout=None, auth=None):
             return self._create_client(connector=connector, headers=headers, timeout=timeout)
 
         try:
-            async with (
-                streamablehttp_client(str(connector.url), httpx_client_factory=client_factory) as (read, write, _),
-                ClientSession(read, write) as session,
-            ):
+            async with AsyncExitStack() as stack:
+                read, write, _ = await stack.enter_async_context(
+                    streamablehttp_client(
+                        (
+                            f"{await stack.enter_async_context(self._stdio_gateway(connector=connector, preset=preset))}/mcp"
+                            if (preset := self._find_preset(url=connector.url))
+                            and str(preset.url).startswith("mcp+stdio://")
+                            else str(connector.url)
+                        ),
+                        httpx_client_factory=client_factory,
+                    )
+                )
+                session = await stack.enter_async_context(ClientSession(read, write))
                 await session.initialize()
         except ExceptionGroup as excgroup:
             if len(excgroup.exceptions) == 1:
@@ -380,6 +397,7 @@ def client_factory(headers=None, timeout=None, auth=None):
 
     async def mcp_proxy(self, *, connector_id: UUID, request: Request, user: User | None = None):
         connector = await self.read_connector(connector_id=connector_id, user=user)
+        preset = self._find_preset(url=connector.url)
 
         forward_headers = {
             key: request.headers[key]
@@ -389,17 +407,24 @@ async def mcp_proxy(self, *, connector_id: UUID, request: Request, user: User |
 
         exit_stack = AsyncExitStack()
         try:
+            target_url = (
+                f"{await exit_stack.enter_async_context(self._stdio_gateway(connector=connector, preset=preset))}/mcp"
+                if preset and str(preset.url).startswith("mcp+stdio://")
+                else str(connector.url)
+            )
             response = await exit_stack.enter_async_context(
                 self._proxy_client.stream(
                     request.method,
-                    str(connector.url),
+                    target_url,
                     headers=forward_headers
                     | (
                         {"authorization": f"Bearer {connector.auth.token.access_token}"}
                         if connector.state == ConnectorState.connected
                         and connector.auth
                         and connector.auth.token
                         and connector.auth.token.token_type == "bearer"
+                        else {"authorization": f"Bearer {preset.bearer_token}"}
+                        if preset and preset.bearer_token
                         else {}
                     ),
                     content=request.stream(),
@@ -418,6 +443,113 @@ async def stream_fn():
             await exit_stack.pop_all().aclose()
             raise
 
+    @asynccontextmanager
+    async def _stdio_gateway(self, *, connector: Connector, preset: ConnectorPreset):
+        log_tasks: list[asyncio.Task] = []
+        process = None
+        try:
+            namespace = self._configuration.connector.runtime.namespace or self._configuration.k8s_namespace
+            kubeconfig = self._configuration.connector.runtime.kubeconfig or self._configuration.k8s_kubeconfig
+            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+                sock.bind(("127.0.0.1", 0))
+                port = sock.getsockname()[1]
+
+            process = await asyncio.create_subprocess_exec(
+                "supergateway",
+                "--stdio",
+                shlex.join(
+                    [
+                        "kubectl",
+                        "run",
+                        f"conn-{connector.id.hex[:6]}-{token_hex(3)}"[:63],
+                        "--rm=true",
+                        "--restart=Never",
+                        "--attach=true",
+                        "--stdin=true",
+                        "--tty=false",
+                        "--image",
+                        preset.stdio.image,
+                        "--image-pull-policy",
+                        "IfNotPresent",
+                        *(["--namespace", namespace] if namespace else []),
+                        *(["--kubeconfig", str(kubeconfig)] if kubeconfig else []),
+                        *(f"--env={key}={val}" for key, val in preset.stdio.env.items()),
+                        *(["--command", "--", *preset.stdio.command] if preset.stdio.command else ["--"]),
+                        *(preset.stdio.args or []),
+                    ]
+                ),
+                "--outputTransport",
+                "streamableHttp",
+                "--port",
+                str(port),
+                "--streamableHttpPath",
+                "/mcp",
+                "--logLevel",
+                "info",
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+            )
+            if process.stdout:
+                log_tasks.append(
+                    asyncio.create_task(
+                        self._log_process_stream(process.stdout, logging.INFO, f"connector[{connector.id}]")
+                    )
+                )
+            if process.stderr:
+                log_tasks.append(
+                    asyncio.create_task(
+                        self._log_process_stream(process.stderr, logging.WARNING, f"connector[{connector.id}]")
+                    )
+                )
+            await self._wait_for_port(
+                port=port,
+                timeout_seconds=self._configuration.connector.runtime.startup_timeout_seconds,
+                process=process,
+            )
+            yield f"http://127.0.0.1:{port}"
+        finally:
+            for task in log_tasks:
+                task.cancel()
+            if process and process.returncode is None:
+                process.terminate()
+                async with asyncio.timeout(5):
+                    await process.wait()
+                if process.returncode is None:
+                    process.kill()
+
+    async def _log_process_stream(self, stream: asyncio.StreamReader, level: int, prefix: str):
+        try:
+            while line := await stream.readline():
+                logger.log(level, "%s %s", prefix, line.decode("utf-8", errors="replace").rstrip())
+        except asyncio.CancelledError:
+            pass
+
+    async def _wait_for_port(
+        self, *, port: int, timeout_seconds: int, process: asyncio.subprocess.Process | None = None
+    ) -> None:
+        loop = asyncio.get_running_loop()
+        deadline = loop.time() + timeout_seconds
+        while True:
+            if process and process.returncode is not None:
+                raise PlatformError(
+                    "Failed to start stdio connector",
+                    status_code=status.HTTP_502_BAD_GATEWAY,
+                )
+            try:
+                _reader, writer = await asyncio.open_connection("127.0.0.1", port)
+            except OSError as err:
+                if loop.time() >= deadline:
+                    raise PlatformError(
+                        "Timed out while starting stdio connector",
+                        status_code=status.HTTP_504_GATEWAY_TIMEOUT,
+                    ) from err
+                await asyncio.sleep(0.1)
+                continue
+            else:
+                writer.close()
+                await writer.wait_closed()
+                return
+
 
 @alru_cache(ttl=timedelta(days=1).seconds)
 async def _register_client(resource_server_url: str, *, redirect_uri: str) -> _ClientRegistrationResponse:

helm/values.yaml

@@ -243,6 +243,9 @@ affinity: { }
 uvicornTimeoutKeepAlive: 5
 
 connector:
+  runtime:
+    namespace: ""
+    startupTimeoutSeconds: 60
   presets:
     - url: "https://mcp.stripe.com"
       metadata:
@@ -260,6 +263,14 @@ connector:
       metadata:
         name: "GitHub"
         description: "Access and interact with your GitHub repositories and code intelligence"
+    - url: "mcp+stdio://filesystem"
+      metadata:
+        name: "Filesystem"
+        description: "Run the stdio filesystem MCP server inside the cluster"
+      stdio:
+        image: "ghcr.io/modelcontextprotocol/python-servers/filesystem:latest"
+        args:
+          - "--root=/data"
 ui:
   replicaCount: 1
   enabled: true