How to Set Up AI Agent Dagster Storage

Dagster storage for AI agents handles persistence of assets generated by agent-driven pipelines. These include model outputs, intermediate datasets, embeddings, and execution metadata.

graph TD
  A[AI Agent Pipeline] --> B[Dagster Op/Asset]
  B --> C[IO Manager]
  C --> D[Fastio Workspace]
  D --> E[MCP Tools]
  E --> F[RAG Query]
  F --> G[Agent Response]

This architecture separates compute from storage. Agents produce assets; Dagster materializes them to Fastio. Intelligence Mode auto-indexes files for semantic search.

Fastio differs from S3 by offering agent-native tools. Upload via REST API or MCP, query with citations, transfer ownership to humans.

Helpful references: Fastio Workspaces, Fastio Collaboration, and Fastio AI.

AI agents in Dagster require durable storage for retries, parallelism, and observability. Without persistence, failures lose artifacts, halting workflows.

Key reasons:

• Retry Safety: Re-execute failed ops without recomputing upstream.
• Multi-Agent Coordination: Share assets across agents via workspaces.
• Cost Control: Reuse embeddings/models instead of regenerating.
• Human Review: Transfer workspaces to teams for validation.

Dagster powers AI/ML at scale. Teams use it for data prep, fine-tuning, and inference pipelines.

Persistent Storage for Dagster Agents?

MCP-compatible workspaces, built-in RAG with Ripley, and 19 named-mode tools for Dagster pipeline assets.

Start 14-Day Trial

Implement a custom IO manager to write assets to Fastio. Small assets go in one multipart POST to https://api.fast.io/current/upload/. Create an API key in Settings > Devices & Agents > API Keys, or with POST /current/user/auth/key/. Workspace IDs are 19-digit numeric strings.

from dagster import Definitions
import requests

class FastIOManager:
    def __init__(self, workspace_id: str, api_key: str):
        self.workspace_id = workspace_id
        self.api_key = api_key

def write(self, context, obj):
        filename = "asset.bin"
        resp = requests.post(
            "https://api.fast.io/current/upload/",
            headers={"Authorization": f"Bearer {self.api_key}"},
            files={"chunk": (filename, obj)},
            data={
                "name": filename,
                "size": str(len(obj)),
                "action": "create",
                "instance_id": self.workspace_id,
                "folder_id": "root",
            },
        )
        return resp.json()["new_file_id"]

Load in definitions:

defs = Definitions(
    assets=[my_agent_asset],
    resources={"io_manager": FastIOManager(
        workspace_id="1234567890123456789",
        api_key="your_key",
    )}
)

A successful small upload returns HTTP 201 with result, id, and new_file_id. Large assets open a session on the same /current/upload/ route (omit chunk), then POST /current/upload/{id}/chunk/?order=N&size=N, POST /current/upload/{id}/complete/, and GET /current/upload/{id}/details/?wait=60. Agents in the same pipeline can also call MCP at https://mcp.fast.io/mcp.

Fastio's MCP server exposes 19 named-mode tools over Streamable HTTP at https://mcp.fast.io/mcp (use https://mcp.fast.io/mcp/key with a Bearer header; legacy SSE is https://mcp.fast.io/sse). Dagster ops call those tools with JSON-RPC.

Example MCP call in a Dagster op:

import os
import requests

MCP_URL = "https://mcp.fast.io/mcp/key"
HEADERS = {
    "Authorization": f"Bearer {os.environ['FASTIO_API_KEY']}",
    "Content-Type": "application/json",
}

payload = {
    "jsonrpc": "2.0",
    "id": 1,
    "method": "tools/call",
    "params": {
        "name": "upload",
        "arguments": {
            "action": "web-import",
            "url": "https://example.com/report.pdf",
            "profile_type": "workspace",
            "profile_id": os.environ["FASTIO_WORKSPACE_ID"],
        },
    },
}
response = requests.post(MCP_URL, headers=HEADERS, json=payload)

Use the storage tool (list, search, details) to inspect workspace files, and ai with action ask for a cited answer from Ripley. Watch file activity with GET /current/events/search/ or GET /current/activity/poll/{entityId}?wait=95&lastactivity={timestamp}.

Unique gap-filler: No other storage offers MCP-native Dagster integration.

For dagster pipelines agents:

• Use file locks: Acquire with POST /current/workspace/{workspace_id}/storage/{node_id}/lock/, keep the lock with POST /current/workspace/{workspace_id}/storage/{node_id}/lock/heartbeat/, and release with DELETE /current/workspace/{workspace_id}/storage/{node_id}/lock/.
• Activity: Poll GET /current/activity/poll/{entityId}?wait=95&lastactivity={timestamp} or search GET /current/events/search/ to advance downstream ops when files change.
• RAG: Query indexed assets in Intelligence Mode. Ripley answers through MCP ai (ask) or POST /current/workspace/{workspace_id}/ai/agent/.
• Ownership Transfer: Agent builds pipeline outputs, hands to human.

Best practices:

1. Partition assets by run ID.
2. Use metadata for lineage.
3. Monitor via Dagster UI + Fastio audit logs.

Edge cases: Large files use POST /current/upload/ to open a session, then POST /current/upload/{id}/chunk/?order=N&size=N, POST /current/upload/{id}/complete/, and GET /current/upload/{id}/details/?wait=60.

Common problems:

• Auth Failures: Verify API keys in Dagster config. Create a key in Settings > Devices & Agents > API Keys, or POST /current/user/auth/key/.
• Usage: Check GET /current/org/{org_id}/billing/details/ and GET /current/org/{org_id}/billing/usage/meters/list/.
• Concurrency: Use file locks for multi-agent writes with POST /current/workspace/{workspace_id}/storage/{node_id}/lock/.

Test pipeline:

dagster dev -f dagster_dagster.py

Check Fastio workspace for assets. List a folder with GET /current/workspace/{workspace_id}/storage/{parent_id}/list/.