How to Integrate Fastio API with CrewAI Workflows

Multi-agent CrewAI workflows often need reliable file sharing. Agents create reports, datasets, and assets. These files must survive restarts and work across agents. Local storage disappears on restart. Basic options lack coordination between agents.

Fastio workspaces offer API access to versioned files and locks for concurrent use. They also include AI indexing. Agents share files without polling or running extra servers. Locks block overwrites in parallel work.

Few guides explain file sharing for multiple CrewAI agents in detail.

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

Enable Shared Storage for CrewAI Agents

Start with 50GB free, 5,000 credits/month. Agents use 251 MCP tools for full control. Built for integrate fast api with crewai workflows workflows.

Get Free Agent Tier

Install CrewAI: pip install crewai.

Sign up for a Fastio agent account at fast.io (no credit card needed). Generate an API key at Settings > API Keys.

Set FASTIO_API_KEY and use base URL https://api.fast.io/current/.

Check Fastio docs for endpoints like /storage/list, /upload/initiate.

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"]

Extend BaseTool to handle uploads, lists, and locks.

from crewai_tools import BaseTool
import requests
import os

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

def _run(self, action: str, **kwargs) -> str:
        headers = {"Authorization": f"Bearer {os.getenv('FASTIO_API_KEY')}"}
        base_url = "https://api.fast.io/current"
        workspace_id = os.getenv('FASTIO_WORKSPACE_ID')  # Set from Step 1

if action == "list_files":
            resp = requests.get(f"{base_url}/workspaces/{workspace_id}/storage/root/?limit=20", headers=headers)
            files = [f['name'] for f in resp.json().get('response', [])]
            return f"Files: {', '.join(files)}"

elif action == "upload_file":
            file_path = kwargs['file_path']
            file_name = os.path.basename(file_path)
            with open(file_path, 'rb') as f:
                files = {'file': (file_name, f)}
                resp = requests.post(f"{base_url}/workspaces/{workspace_id}/storage/root/upload/", files=files, headers=headers)
            return f"Uploaded {file_name}: {resp.json()}"

elif action == "lock_file":
            node_id = kwargs['node_id']
            resp = requests.post(f"{base_url}/workspaces/{workspace_id}/storage/{node_id}/lock/", headers=headers)
            return f"Locked {node_id}: {resp.json()}"

raise ValueError(f"Unknown action: {action}")

Add more actions as needed, like AI queries or creating shares.

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.