How to Integrate Fastio API with CrewAI Workflows

Fastio workspaces offer API access to versioned files and granular permissions for concurrent use. Once Intelligence is enabled for the workspace, files are indexed for semantic search and Ripley AI chat. Version history and audit logs prevent overwrites in parallel work.

Enable Shared Storage for CrewAI Agents

Shared persistent storage for CrewAI agents. Agents use a consolidated toolset for full workspace control.

Start Business Trial

• How to Integrate the Fastio API with n8n WorkflowsGuide to integrate fast api with n8n workflows: Connect n8n workflows to Fastio's API to automate file storage and...
• How to Integrate Fastio API with LangChain ToolsConnect Fastio's API to LangChain tools. AI agents then get lasting file storage, RAG across multiple files, and...
• LangGraph vs CrewAI: Which Multi-Agent Framework to Choose in 2026LangGraph and CrewAI are the two most-searched multi-agent frameworks heading into 2026. This comparison goes beyond...
• How to Deploy CrewAI to ProductionDeploying CrewAI crews to production moves notebook experiments to reliable systems. Notebooks suit tests, but lack...
• OpenAI Agents SDK vs CrewAI: Choosing the Right Agent FrameworkOpenAI Agents SDK and CrewAI solve multi-agent orchestration in fundamentally different ways. This comparison breaks...
• Best Tools for CrewAI Agents: Top Picks for 2026CrewAI agents need good tools to be useful. The framework handles coordination, but external integrations let agents...

More on this subject: Multi-Agent Systems (54 guides)

Extend BaseTool to handle uploads, listings, and version history.

import os
import requests
from crewai_tools import BaseTool

class FastioTool(BaseTool):
    name: str = "Fastio Workspace Tool"
    description: str = "Interact with Fastio workspace: upload, list, and inspect versions for CrewAI agents."

def _run(self, action: str, **kwargs) -> str:
        api_key = os.getenv("FASTIO_API_KEY")
        base_url = "https://api.fast.io/current"
        headers = {"Authorization": f"Bearer {api_key}"}
        workspace_id = kwargs.get("workspace_id")
        node_id = kwargs.get("node_id")

if action == "upload":
            return f"Uploaded {kwargs.get('file_path')}"
        elif action == "list_files":
            resp = requests.get(f"{base_url}/workspaces/{workspace_id}/storage/", headers=headers)
            return f"Files: {resp.json()}"
        elif action == "get_versions":
            resp = requests.get(f"{base_url}/workspaces/{workspace_id}/storage/{node_id}/versions/", headers=headers)
            return f"Versions for {node_id}: {resp.json()}"
        return "Unknown action"

Use Bearer token authentication. Store FASTIO_API_KEY as an environment variable.

Test the connection:

import requests
headers = {"Authorization": f"Bearer {os.getenv('FASTIO_API_KEY')}"}
response = requests.get("https://api.fast.io/current/user/", headers=headers)
print(response.json())

The response shows your user and org details. Create a workspace with POST /current/org/{org_id}/workspaces/.

org_data = {"domain": "my-crewai-agent", "name": "CrewAI Workspace Org"}
org_resp = requests.post("https://api.fast.io/current/orgs/", json=org_data, headers=headers)
org_id = org_resp.json()["id"]
ws_data = {"folder_name": "crewai-workflow", "name": "CrewAI Shared Files"}
ws_resp = requests.post(f"https://api.fast.io/current/orgs/{org_id}/workspaces/", json=ws_data, headers=headers)
workspace_id = ws_resp.json()["id"]

In multi-agent architectures, automatic file version history and granular permissions coordinate parallel access safely without brittle advisory locks. When agents update shared files, Fastio retains all versions so changes can be audited or restored.

Instead of webhooks, Fastio provides a realtime activity feed you can poll and a WebSocket events feed to notify agents of file modifications instantly.

Set up your agents and crew.

from crewai import Agent, Task, Crew
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")

researcher = Agent(
    role="Researcher",
    goal="Research topics and save reports to Fastio",
    backstory="Expert researcher using shared storage.",
    llm=llm.
    tools=[FastioTool()],
    verbose=True
)

analyst = Agent(
    role="Analyst",
    goal="Analyze reports from workspace",
    backstory="Data analyst coordinating with team.",
    llm=llm.
    tools=[FastioTool()],
    verbose=True
)

task1 = Task(description="Research AI trends, upload report.", agent=researcher)
task2 = Task(description="Lock report, analyze, save summary.", agent=analyst)

crew = Crew(agents=[researcher, analyst], tasks=[task1, task2])
result = crew.kickoff()

Call crew.kickoff() to run the workflow.

Locks keep parallel access safe. The researcher, for example, locks a file before editing it.

Webhooks alert you to changes. POST to /webhooks/ to subscribe to events.

Enable intelligence mode for RAG queries across workspace files.

Example query code:

### Add to tool
elif action == "query_files":
    scope = kwargs.get('scope', 'root')
    resp = requests.post(f"{base_url}/workspaces/{workspace_id}/ai/chat/", 
                         data={'type': 'chat_with_files', 'query_text': kwargs['question'], 'folders_scope': scope},
                         headers=headers)
    return resp.json()['messages'][-1]['text']

Agents share file state this way.

Define clear tool contracts and fallback behavior so agents fail safely when dependencies are unavailable. This improves reliability in production workflows.

For rate limits, use pagination and retries.

Chunk large file uploads.

Monitor credits: GET /org/{id}/billing/usage/.

Test in local setup before going to production.