# How to Handle Fastio Realtime Events with Python FastAPI

Handle Fastio real-time file updates in FastAPI using WebSocket event feeds or activity polling. Build reactive AI agent workflows with asynchronous background tasks.

Source: https://fast.io/resources/implement-fastio-webhooks-fastapi/
Last reviewed: 2026-02-23

## Why Choose FastAPI for Fastio Realtime Events?

Real-time event feeds are the foundation of modern file automation. When someone uploads or modifies a file in a Fastio workspace, your system needs to know right away so it can trigger an AI agent, update a database, or notify human collaborators. Instead of webhooks, Fastio provides a WebSocket events feed and a realtime activity feed that can be polled via the REST API. Python is the primary language for AI engineering, making FastAPI a natural choice for consuming these event feeds.

According to the FastAPI Documentation, adopting the framework increases feature development speed by 200% to 300%. That speed helps when building event-driven consumers. FastAPI handles asynchronous connections cleanly, allowing you to maintain persistent WebSocket connections or run periodic polling tasks in the background without blocking your web API.

FastAPI uses standard Python types and Pydantic models for structured data validation. Instead of writing boilerplate code to parse raw event objects, you define typed models and let the framework validate incoming payloads. You can then focus on building your custom [AI agent integration](/storage-for-agents/) or media pipeline.

Helpful references: [Fastio Workspaces](/product/workspaces/), [Fastio Collaboration](/product/collaboration/), and [Fastio AI](/product/ai/).

## What to check before scaling Fastio event handling with Python FastAPI

Before writing code, understand how Fastio distributes event data. Fastio exposes a WebSocket events feed for real-time push streaming and an activity feed via the REST API at https://api.fast.io/current/ that can be polled.

Each event payload includes metadata about the resource, the user or agent that triggered the action, and the workspace where the event occurred. A file creation event, for example, includes the file ID, name, size, workspace ID, and timestamp.

This context is essential when integrating AI tools. If you configure an agent to process new documents, the event payload tells it exactly which file to inspect and which workspace contains it. Fastio workspaces are shared environments for agents and humans, with full version history and an append-only audit log tracking every update.

To scale reliably, validate event structures before passing them to downstream workers. We use Pydantic models to enforce strict schemas before processing event data.

## Defining the Pydantic Event Model

Handling incoming event data in FastAPI starts with a strict Pydantic model. This ensures your application validates event structures reliably.

Here is an example Pydantic model for parsing Fastio workspace events:

```python
from pydantic import BaseModel
from typing import Dict, Any, Optional
from datetime import datetime

class EventUser(BaseModel):
    id: str
    email: Optional[str] = None

class EventResource(BaseModel):
    id: str
    type: str
    name: str
    size: Optional[int] = None

class FastioEventPayload(BaseModel):
    event_id: str
    event_type: str
    workspace_id: str
    created_at: datetime
    resource: EventResource
    user: Optional[EventUser] = None
    metadata: Dict[str, Any] = {}
```

This nested structure maps directly to Fastio workspace events. Typing `payload.event_type` in your IDE provides autocompletion and type checking, keeping your event ingestion logic clean and maintainable.

## Connecting to the Fastio WebSocket Events Feed

Fastio provides a WebSocket events feed that pushes real-time workspace updates directly to connected clients. This eliminates the need to expose a public inbound URL or configure webhook endpoints.

Here is how you can connect to the event feed using Python and FastAPI:

```python
import asyncio
import json
import os
import websockets
from fastapi import FastAPI, BackgroundTasks

app = FastAPI()
FASTIO_API_KEY = os.environ.get("FASTIO_API_KEY", "")
FASTIO_WS_URL = "wss://api.fast.io/current/events"

async def listen_to_fastio_events():
    headers = {"Authorization": f"Bearer {FASTIO_API_KEY}"}
    async with websockets.connect(FASTIO_WS_URL, extra_headers=headers) as ws:
        while True:
            message = await ws.recv()
            event_data = json.loads(message)
            print(f"Received Fastio event: {event_data.get('event_type')}")
            # Process event asynchronously

@app.on_event("startup")
async def startup_event():
    asyncio.create_task(listen_to_fastio_events())
```

Using an outbound WebSocket connection ensures your consumer runs securely inside private networks without opening inbound firewall ports.

## Handling File Events Asynchronously

After receiving an event, your application processes the file. In FastAPI, you can use `BackgroundTasks` or an async task queue to handle heavy operations like AI analysis, OCR, or document summarization.

Here is an implementation example:

```python
def process_file_event(payload: FastioEventPayload):
    ### This function runs in the background
    if payload.event_type == "file.created":
        print(f"Processing uploaded file: {payload.resource.name}")
        ### Invoke MCP tools or call Fastio REST API to retrieve and process file

@app.post("/events/process")
async def handle_event(
    payload: FastioEventPayload,
    background_tasks: BackgroundTasks
):
    background_tasks.add_task(process_file_event, payload)
    return {"status": "queued", "event_id": payload.event_id}
```

This pattern keeps your event ingestion decoupled from long-running agent tasks.

## Testing Event Consumers Locally

Testing event consumers locally is straightforward because Fastio relies on outbound WebSocket connections and REST API polling rather than inbound webhooks.

Because your local server connects outbound to Fastio, you do not need tunneling tools or public IP addresses. Start your FastAPI server on localhost, provide your scoped API key, and upload test files through the Fastio web dashboard.

Your local terminal will immediately log the event payloads as they arrive over the WebSocket feed, making local development fast and secure.

## Integrating Event Feeds with the Business Trial

Real-time event feeds enable responsive AI automation. Fastio gives developers the infrastructure to build these multi-agent workflows.

Fastio offers a 14-day Business Trial requiring a credit card. You can use this trial to build and test reactive event workflows. Set up a workspace, connect your event handler to your FastAPI app, and explore automated file processing.

Streaming file events turns regular cloud storage into an intelligent workspace. Instead of manually checking for updates, your FastAPI application reacts immediately to workspace activity.

## Frequently asked questions

### How does Fastio notify applications of workspace events?

Fastio does not use webhooks. Instead, Fastio provides a WebSocket events feed for real-time push streaming and a realtime activity feed that can be polled via the REST API.

### How do you listen to Fastio events in Python FastAPI?

In FastAPI, you can run an asynchronous background task during application startup that connects to the Fastio WebSocket events feed or periodically polls the activity feed using an API key.

### Do I need a public URL or ngrok to test Fastio events?

No. Because your FastAPI application initiates an outbound WebSocket connection or REST API polling request to Fastio, you do not need public tunneling tools like ngrok to test locally.

### How do I authenticate with Fastio event feeds?

Authenticate using a scoped, long-lived API key granted by a human administrator or through PKCE browser login, passing the bearer token in the authorization header.

### What happens if the WebSocket connection disconnects?

Implement an exponential backoff reconnection loop. Upon reconnecting, you can query Fastio's realtime activity feed or audit log to catch up on any events that occurred during downtime.

## About Fast.io

Fast.io provides shared workspaces where people and AI agents work on the same files, with built-in semantic search and citation-backed chat over what they hold. Agents reach it through a remote MCP server at https://mcp.fast.io/mcp, a REST API at https://api.fast.io/current/, and a command line client published on npm as @vividengine/fastio-cli.