Skip to content

Artifact loading

octomate.artifacts

Authorized artifact reads, mounted by the host independently of any channel.

GET /api/artifacts/{artifact_id} reads an owned artifact without requiring a thread. GET /api/trunkline/threads/{thread_id}/artifacts/{message_id} additionally requires thread access and a visible ledger reference, including fork cutoffs.

Both return the saved invocation, a freshly fetched UI resource, and the configured sandbox origin with Cache-Control: no-store. No provider credentials are returned; there is no caller-selected resource URI or tool execution endpoint. HTML stays untrusted JSON data here. The SPA viewer uses the configured artifact prefix; GET /api/artifacts/config supplies that public routing configuration. The isolated runtime is served on its configured origin through the same listener.

The proxy's response CSP bounds the inner document's inherited policy. A second, stricter CSP inside srcdoc removes the proxy's same-origin frame allowance. The inner iframe grants scripts only, so it has an opaque origin and cannot navigate the console, open popups, submit forms, or download files. No browser permissions are granted. The policy URL is public and contains no artifact or invocation data.

logger module-attribute 

logger = logging.getLogger(__name__)

NO_STORE module-attribute 

NO_STORE = {'Cache-Control': 'no-store'}

ASSETS module-attribute 

ASSETS = Path(__file__).parent / 'static' / 'artifacts'

PROXY_HTML module-attribute 

PROXY_HTML = (ASSETS / 'sandbox.html').read_text()

PERMISSIONS module-attribute 

PERMISSIONS = "camera=(), microphone=(), geolocation=(), clipboard-read=(), clipboard-write=(), fullscreen=(), display-capture=(), payment=(), usb=()"

artifact_router module-attribute 

artifact_router = APIRouter(
    tags=["artifacts"],
    dependencies=[Depends(browser_request)],
)

ArtifactSandbox

Dispatch the configured sandbox origin before any console or account route.

Reverse proxies must preserve Host and the external scheme. Requests on this origin cannot reach Octomate APIs, OpenAPI, cookies, WebSockets, or SPA assets. The sandbox paths are unavailable on all other origins.

Source code in octomate/artifacts.py 
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
class ArtifactSandbox:
    """Dispatch the configured sandbox origin before any console or account route.

    Reverse proxies must preserve Host and the external scheme. Requests on this
    origin cannot reach Octomate APIs, OpenAPI, cookies, WebSockets, or SPA assets.
    The sandbox paths are unavailable on all other origins.
    """

    def __init__(self, app: ASGIApp, *, origin: AnyHttpUrl | None) -> None:
        self.app: ASGIApp = app
        self.origin: AnyHttpUrl | None = origin
        self.router: ASGIApp = ExceptionMiddleware(
            Router(
                routes=[
                    Route("/artifacts/sandbox", sandbox_proxy),
                    Route("/artifacts/sandbox.js", sandbox_script),
                ]
            )
        )

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] not in {"http", "websocket"}:
            await self.app(scope, receive, send)
            return
        request = Request({**scope, "type": "http"})
        url = AnyHttpUrl(
            str(request.base_url)
            .replace("ws://", "http://")
            .replace("wss://", "https://")
        )
        isolated = (
            self.origin is not None
            and url.host == self.origin.host
            and (
                url.port == self.origin.port
                or (url.port in {80, 443} and self.origin.port in {80, 443})
            )
        )
        if (
            scope["type"] == "http"
            and isolated
            and self.origin is not None
            and url.scheme == self.origin.scheme
        ):
            await self.router(scope, receive, send)
        elif isolated or scope["path"] in {
            "/artifacts/sandbox",
            "/artifacts/sandbox.js",
        }:
            if scope["type"] == "websocket":
                await WebSocket(scope, receive, send).close()
            else:
                await Response(status_code=404)(scope, receive, send)
        else:
            await self.app(scope, receive, send)

app instance-attribute 

app: ASGIApp = app

origin instance-attribute 

origin: AnyHttpUrl | None = origin

router instance-attribute 

router: ASGIApp = ExceptionMiddleware(
    Router(
        routes=[
            Route("/artifacts/sandbox", sandbox_proxy),
            Route("/artifacts/sandbox.js", sandbox_script),
        ]
    )
)

artifact_errors 

artifact_errors() -> Generator[None]

Translate artifact loading failures into safe, uncached HTTP responses.

Unknown or inaccessible artifacts/installations return 404, missing or same-origin sandbox configuration returns 503, and invalid or failed upstream resource reads return 502. Upstream exception details stay server-side.

Source code in octomate/artifacts.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@contextmanager
def artifact_errors() -> Generator[None]:
    """Translate artifact loading failures into safe, uncached HTTP responses.

    Unknown or inaccessible artifacts/installations return 404, missing or
    same-origin sandbox configuration returns 503, and invalid or failed upstream
    resource reads return 502. Upstream exception details stay server-side.
    """
    try:
        yield
    except ArtifactUnavailable as error:
        raise HTTPException(404, str(error), headers=NO_STORE) from error
    except ArtifactSandboxUnavailable as error:
        raise HTTPException(503, str(error), headers=NO_STORE) from error
    except (ValueError, httpx2.HTTPError, MCPError, TimeoutError) as error:
        logger.warning("Failed to load artifact (%s)", type(error).__name__)
        raise HTTPException(
            502, "MCP app resource could not be loaded", headers=NO_STORE
        ) from error

artifact_config async 

artifact_config(
    manager: Annotated[
        ArtifactManager, Depends(artifact_manager)
    ],
) -> ArtifactConfig

Public viewer routing settings; provider credentials are never exposed.

Source code in octomate/artifacts.py 
83
84
85
86
87
88
@artifact_router.get("/api/artifacts/config", response_model=ArtifactConfig)
async def artifact_config(
    manager: Annotated[ArtifactManager, Depends(artifact_manager)],
) -> ArtifactConfig:
    """Public viewer routing settings; provider credentials are never exposed."""
    return manager.config

read_artifact async 

read_artifact(
    artifact_id: UUID7,
    user: Annotated[User, Depends(current_user)],
    manager: Annotated[
        ArtifactManager, Depends(artifact_manager)
    ],
    request: Request,
) -> McpAppView

Read owned content independently of whether any thread still references it.

Source code in octomate/artifacts.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
@artifact_router.get(
    "/api/artifacts/{artifact_id}",
    response_model=McpAppView,
    response_model_exclude_none=True,
)
async def read_artifact(
    artifact_id: UUID7,
    user: Annotated[User, Depends(current_user)],
    manager: Annotated[ArtifactManager, Depends(artifact_manager)],
    request: Request,
) -> McpAppView:
    """Read owned content independently of whether any thread still references it."""
    with artifact_errors():
        return await manager.load(
            artifact_id, user, console_origin=AnyHttpUrl(str(request.base_url))
        )

read_thread_artifact async 

read_thread_artifact(
    thread_id: UUID7,
    message_id: UUID7,
    user: Annotated[User, Depends(current_user)],
    manager: Annotated[
        ArtifactManager, Depends(artifact_manager)
    ],
    request: Request,
) -> McpAppView

Read an owned artifact only through a reference visible in this thread.

Source code in octomate/artifacts.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
@artifact_router.get(
    "/api/trunkline/threads/{thread_id}/artifacts/{message_id}",
    response_model=McpAppView,
    response_model_exclude_none=True,
)
async def read_thread_artifact(
    thread_id: UUID7,
    message_id: UUID7,
    user: Annotated[User, Depends(current_user)],
    manager: Annotated[ArtifactManager, Depends(artifact_manager)],
    request: Request,
) -> McpAppView:
    """Read an owned artifact only through a reference visible in this thread."""
    with artifact_errors():
        return await manager.load_from_message(
            thread_id,
            message_id,
            user,
            console_origin=AnyHttpUrl(str(request.base_url)),
        )

sandbox_csp 

sandbox_csp(
    policy: ArtifactSandboxPolicy, *, proxy: bool
) -> str

Construct restrictive source lists; only the outer proxy may frame itself.

Source code in octomate/artifacts.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def sandbox_csp(policy: ArtifactSandboxPolicy, *, proxy: bool) -> str:
    """Construct restrictive source lists; only the outer proxy may frame itself."""
    csp = policy.csp
    resources = " ".join(csp.resource_domains or [])
    frames = " ".join((["'self'"] if proxy else []) + (csp.frame_domains or []))
    directives = [
        "default-src 'none'",
        f"script-src 'self' 'unsafe-inline' {resources}",
        f"style-src 'self' 'unsafe-inline' {resources}",
        f"img-src 'self' data: {resources}",
        f"font-src 'self' data: {resources}",
        f"media-src 'self' data: {resources}",
        "connect-src " + (" ".join(csp.connect_domains or []) or "'none'"),
        "frame-src " + (frames or "'none'"),
        "base-uri " + (" ".join(csp.base_uri_domains or []) or "'none'"),
        "object-src 'none'",
        "form-action 'none'",
        "worker-src 'none'",
    ]
    if proxy:
        directives.append(
            "frame-ancestors " + str(policy.parent_origin).removesuffix("/")
        )
    return "; ".join(directive.rstrip() for directive in directives)

sandbox_proxy async 

sandbox_proxy(request: Request) -> Response

Validate policy before emitting headers; HTML arrives later via postMessage.

Source code in octomate/artifacts.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
async def sandbox_proxy(request: Request) -> Response:
    """Validate policy before emitting headers; HTML arrives later via postMessage."""
    raw = request.query_params.get("policy", "")
    if len(raw) > 16384:
        return Response(status_code=422)
    try:
        policy = ArtifactSandboxPolicy.model_validate_json(raw)
    except ValidationError:
        return Response(status_code=422)
    if policy.parent_origin == AnyHttpUrl(str(request.base_url)):
        return Response(status_code=422)
    html = PROXY_HTML.replace(
        "{{parent_origin}}",
        escape(str(policy.parent_origin).removesuffix("/"), quote=True),
    ).replace("{{csp}}", escape(sandbox_csp(policy, proxy=False), quote=True))
    return HTMLResponse(
        html,
        headers={
            "Cache-Control": "no-store",
            "Content-Security-Policy": sandbox_csp(policy, proxy=True),
            "Permissions-Policy": PERMISSIONS,
            "Referrer-Policy": "no-referrer",
            "X-Content-Type-Options": "nosniff",
        },
    )

sandbox_script async 

sandbox_script(request: Request) -> Response

The proxy's bundled script, with no console assets on this origin.

Source code in octomate/artifacts.py 
184
185
186
187
188
189
190
async def sandbox_script(request: Request) -> Response:
    """The proxy's bundled script, with no console assets on this origin."""
    return FileResponse(
        ASSETS / "sandbox.js",
        media_type="text/javascript",
        headers={"Cache-Control": "no-store", "X-Content-Type-Options": "nosniff"},
    )

octomate.managers.artifacts

Owned artifact loading, including access through a thread's visible ledger.

ArtifactUnavailable

Bases: LookupError

The artifact, its ledger reference, or its installation is inaccessible.

Source code in octomate/managers/artifacts.py 
24
25
class ArtifactUnavailable(LookupError):
    """The artifact, its ledger reference, or its installation is inaccessible."""

ArtifactSandboxUnavailable

Bases: ValueError

Artifact loading requires a configured sandbox on a separate origin.

Source code in octomate/managers/artifacts.py 
28
29
class ArtifactSandboxUnavailable(ValueError):
    """Artifact loading requires a configured sandbox on a separate origin."""

ArtifactViewerUnavailable

Bases: ValueError

An absolute artifact link requires a configured viewer origin.

Source code in octomate/managers/artifacts.py 
32
33
class ArtifactViewerUnavailable(ValueError):
    """An absolute artifact link requires a configured viewer origin."""

ArtifactManager

Bases: Manager

Authorize artifact ownership and hosting before fetching saved UI resources.

Source code in octomate/managers/artifacts.py 
 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
class ArtifactManager(Manager):
    """Authorize artifact ownership and hosting before fetching saved UI resources."""

    def __init__(
        self, *, config: ArtifactConfig, mcp: McpManager, threads: ThreadManager
    ) -> None:
        self.config: ArtifactConfig = config
        self.mcp: McpManager = mcp
        self.threads: ThreadManager = threads

    async def overview(self, artifact_id: UUID7, user: User) -> McpAppOverview:
        """Read an owned artifact's card metadata without contacting its provider.

        Only the saved title, provider name, identity, and viewer URL leave this
        method. An overview does not grant access to its recipients or guarantee
        that the installation is still available; the viewer checks that on open.
        Missing viewer configuration raises ArtifactViewerUnavailable.
        """
        async with async_session() as session:
            artifact = await session.one_or_none(
                McpAppArtifact,
                expressions=[
                    McpAppArtifact["id"] == artifact_id,
                    McpAppArtifact["owner_id"] == user.id,
                ],
            )
        if artifact is None:
            raise ArtifactUnavailable("Artifact unavailable")
        origin = self.config.viewer_origin
        if origin is None:
            raise ArtifactViewerUnavailable("Configure artifacts.viewer_origin")
        return McpAppOverview(
            artifact_id=artifact.id,
            title=artifact.title,
            provider_name=artifact.invocation.provider_name,
            viewer_url=AnyHttpUrl(
                f"{str(origin).rstrip('/')}{self.config.path}/{artifact.id}"
            ),
        )

    async def load(
        self, artifact_id: UUID7, user: User, *, console_origin: AnyHttpUrl
    ) -> McpAppView:
        """Read owned content independently of its thread, without executing tools.

        The saved installation must still be owned, enabled, and authorized. A
        separate sandbox origin is required before any UI resource is fetched.
        """
        async with async_session() as session:
            artifact = await session.one_or_none(
                McpAppArtifact,
                expressions=[
                    McpAppArtifact["id"] == artifact_id,
                    McpAppArtifact["owner_id"] == user.id,
                ],
            )
        if artifact is None:
            raise ArtifactUnavailable("Artifact unavailable")
        sandbox = self.config.sandbox_origin
        if sandbox is None or (sandbox.scheme, sandbox.host, sandbox.port) == (
            console_origin.scheme,
            console_origin.host,
            console_origin.port,
        ):
            raise ArtifactSandboxUnavailable(
                "Configure a separate artifact sandbox origin"
            )
        try:
            resource = await self.mcp.app_resource(user, artifact.invocation)
        except McpUnavailable as error:
            raise ArtifactUnavailable("Artifact installation unavailable") from error
        policy = ArtifactSandboxPolicy(
            parent_origin=console_origin,
            csp=resource.ui.csp or McpAppResourceCSP(),
        )
        query = urlencode({"policy": policy.model_dump_json(by_alias=True)})
        return McpAppView(
            artifact=artifact,
            resource=resource,
            sandbox_origin=sandbox,
            sandbox_url=AnyHttpUrl(f"{sandbox}artifacts/sandbox?{query}"),
            viewer_path=f"{self.config.path}/{artifact.id}",
        )

    async def load_from_message(
        self,
        thread_id: UUID7,
        message_id: UUID7,
        user: User,
        *,
        console_origin: AnyHttpUrl,
    ) -> McpAppView:
        """Read an owned artifact through an accessible thread and visible reference.

        Inherited references obey the fork's cutoff. Thread participation does not
        grant content ownership; the standalone load applies that check as well.
        """
        thread = await self.threads.get(thread_id, with_messages=False, user_id=user.id)
        if thread is None:
            raise ArtifactUnavailable("Thread unavailable")
        async with async_session() as session:
            message = await session.one_or_none(
                ThreadArtifact,
                expressions=[ThreadArtifact["id"] == message_id, in_ledger(thread.id)],
                options=[noload("*")],
            )
        if message is None:
            raise ArtifactUnavailable("Artifact reference unavailable")
        return await self.load(message.artifact_id, user, console_origin=console_origin)

config instance-attribute 

config: ArtifactConfig = config

mcp instance-attribute 

mcp: McpManager = mcp

threads instance-attribute 

threads: ThreadManager = threads

overview async 

overview(artifact_id: UUID7, user: User) -> McpAppOverview

Read an owned artifact's card metadata without contacting its provider.

Only the saved title, provider name, identity, and viewer URL leave this method. An overview does not grant access to its recipients or guarantee that the installation is still available; the viewer checks that on open. Missing viewer configuration raises ArtifactViewerUnavailable.

Source code in octomate/managers/artifacts.py 
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
async def overview(self, artifact_id: UUID7, user: User) -> McpAppOverview:
    """Read an owned artifact's card metadata without contacting its provider.

    Only the saved title, provider name, identity, and viewer URL leave this
    method. An overview does not grant access to its recipients or guarantee
    that the installation is still available; the viewer checks that on open.
    Missing viewer configuration raises ArtifactViewerUnavailable.
    """
    async with async_session() as session:
        artifact = await session.one_or_none(
            McpAppArtifact,
            expressions=[
                McpAppArtifact["id"] == artifact_id,
                McpAppArtifact["owner_id"] == user.id,
            ],
        )
    if artifact is None:
        raise ArtifactUnavailable("Artifact unavailable")
    origin = self.config.viewer_origin
    if origin is None:
        raise ArtifactViewerUnavailable("Configure artifacts.viewer_origin")
    return McpAppOverview(
        artifact_id=artifact.id,
        title=artifact.title,
        provider_name=artifact.invocation.provider_name,
        viewer_url=AnyHttpUrl(
            f"{str(origin).rstrip('/')}{self.config.path}/{artifact.id}"
        ),
    )

load async 

load(
    artifact_id: UUID7,
    user: User,
    *,
    console_origin: AnyHttpUrl,
) -> McpAppView

Read owned content independently of its thread, without executing tools.

The saved installation must still be owned, enabled, and authorized. A separate sandbox origin is required before any UI resource is fetched.

Source code in octomate/managers/artifacts.py 
 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
async def load(
    self, artifact_id: UUID7, user: User, *, console_origin: AnyHttpUrl
) -> McpAppView:
    """Read owned content independently of its thread, without executing tools.

    The saved installation must still be owned, enabled, and authorized. A
    separate sandbox origin is required before any UI resource is fetched.
    """
    async with async_session() as session:
        artifact = await session.one_or_none(
            McpAppArtifact,
            expressions=[
                McpAppArtifact["id"] == artifact_id,
                McpAppArtifact["owner_id"] == user.id,
            ],
        )
    if artifact is None:
        raise ArtifactUnavailable("Artifact unavailable")
    sandbox = self.config.sandbox_origin
    if sandbox is None or (sandbox.scheme, sandbox.host, sandbox.port) == (
        console_origin.scheme,
        console_origin.host,
        console_origin.port,
    ):
        raise ArtifactSandboxUnavailable(
            "Configure a separate artifact sandbox origin"
        )
    try:
        resource = await self.mcp.app_resource(user, artifact.invocation)
    except McpUnavailable as error:
        raise ArtifactUnavailable("Artifact installation unavailable") from error
    policy = ArtifactSandboxPolicy(
        parent_origin=console_origin,
        csp=resource.ui.csp or McpAppResourceCSP(),
    )
    query = urlencode({"policy": policy.model_dump_json(by_alias=True)})
    return McpAppView(
        artifact=artifact,
        resource=resource,
        sandbox_origin=sandbox,
        sandbox_url=AnyHttpUrl(f"{sandbox}artifacts/sandbox?{query}"),
        viewer_path=f"{self.config.path}/{artifact.id}",
    )

load_from_message async 

load_from_message(
    thread_id: UUID7,
    message_id: UUID7,
    user: User,
    *,
    console_origin: AnyHttpUrl,
) -> McpAppView

Read an owned artifact through an accessible thread and visible reference.

Inherited references obey the fork's cutoff. Thread participation does not grant content ownership; the standalone load applies that check as well.

Source code in octomate/managers/artifacts.py 
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
async def load_from_message(
    self,
    thread_id: UUID7,
    message_id: UUID7,
    user: User,
    *,
    console_origin: AnyHttpUrl,
) -> McpAppView:
    """Read an owned artifact through an accessible thread and visible reference.

    Inherited references obey the fork's cutoff. Thread participation does not
    grant content ownership; the standalone load applies that check as well.
    """
    thread = await self.threads.get(thread_id, with_messages=False, user_id=user.id)
    if thread is None:
        raise ArtifactUnavailable("Thread unavailable")
    async with async_session() as session:
        message = await session.one_or_none(
            ThreadArtifact,
            expressions=[ThreadArtifact["id"] == message_id, in_ledger(thread.id)],
            options=[noload("*")],
        )
    if message is None:
        raise ArtifactUnavailable("Artifact reference unavailable")
    return await self.load(message.artifact_id, user, console_origin=console_origin)