How to Build a Document Processing Pipeline with Fastio API

A document processing pipeline is an automated system that ingests files, extracts structured data from them, and routes that data to downstream applications. Instead of a person opening a PDF, reading the contents, and manually typing values into a spreadsheet, the pipeline handles every step programmatically.

The typical pipeline follows four stages: upload, trigger, process, and store. A file enters the system through an upload or email attachment. An event fires to notify the processing layer. An AI model or OCR engine reads the document and pulls out specific fields. The structured output gets saved as metadata, pushed to a database, or forwarded to another service.

According to Parseur's 2026 automation guide, organizations that automate document workflows cut manual data entry by 80% or more. The savings compound quickly. DocuExprt estimates that manual document processing costs between $5 and $25 per document when you factor in labor, error correction, and delays. For a team handling hundreds of documents per week, that adds up to thousands of dollars in avoidable overhead.

The challenge for developers is stitching together the infrastructure. A traditional pipeline requires separate services for storage (S3 or Google Cloud Storage), event delivery (Amazon EventBridge or a custom webhook layer), OCR (Tesseract, Google Document AI, or ABBYY), and metadata storage (a relational database or search index). Each integration point is a potential failure, and the whole system needs monitoring.

Fastio collapses several of these layers into one platform. A workspace provides persistent storage. Event feeds deliver file updates in real time. Intelligence Mode auto-indexes documents once enabled for the workspace. And the MCP server gives AI agents direct access to files without downloading them. The result is fewer moving parts and a shorter path from "file uploaded" to "data extracted."

Helpful references: Fastio Workspaces, Fastio AI.

• How to Build an AI Agent Document Redaction PipelineAI agent document redaction automation uses an autonomous agent to detect and remove sensitive information like names,...
• Best Document Processing Tools for AI AgentsDocument processing tools for AI agents automate the extraction, parsing, and transformation of unstructured documents...
• How to Automate Document Processing with AI AgentsAI agent document processing uses autonomous agents to extract, analyze, and transform information from documents...
• How to Build an Agentic File Router with Fastio EventsAn agentic file router uses Fastio realtime events to dispatch uploaded files to specialized AI agents based on...
• How to Build Automated Metadata Extraction Webhook WorkflowsPolling for new files wastes compute and delays processing. Webhook-driven metadata extraction pipelines react to file...
• How to Automate Spreadsheets with AI AgentsManual data entry takes up nearly 30% of the work week. AI agents take over these repetitive tasks by reading,...

More on this subject: Agent File and Document Workflows (183 guides)

Before writing code, it helps to map out the components and data flow. Here is the architecture for a document processing pipeline built on the Fastio API:

1. Upload layer. Documents arrive in a Fastio workspace through the web UI, API upload, MCP tools, or a Receive share (a branded upload portal for external users). Files land in an inbound folder.

2. Event layer. The realtime activity feed or WebSocket events feed fires when a file is created in the inbound folder, delivering the file ID, name, size, MIME type, and workspace context.

3. Routing layer. Your application checks the event payload, validates the file type, and pushes a processing job onto an async queue (Redis, Celery, BullMQ, or a cloud equivalent). Simple text files get lightweight extraction. PDFs and images get routed to AI extraction.

4. Extraction layer. A worker pulls the job, connects an LLM to the Fastio MCP server, and instructs the model to read the document and return structured data. The MCP server handles file access, so the raw document never leaves the Fastio environment.

5. Storage layer. The worker writes the extracted key-value pairs back to the file's custom metadata in Fastio. It moves the file from inbound to processed and updates a status tag. Human reviewers see the results instantly in the same workspace.

This event-driven design decouples ingestion from processing. Your event listener stays fast because it only enqueues a job. The actual extraction happens asynchronously, which means a sudden batch of 200 uploads does not overwhelm your server.

For teams already using Fastio for collaboration, the pipeline runs inside the same workspace where humans review and approve documents. There is no separate "processing bucket" that agents use in isolation. Agents and humans share one environment, which simplifies permissions and audit trails.

Start by creating a dedicated workspace for your pipeline. You can do this through the Fastio dashboard or programmatically via the API.

Create three folders inside the workspace: inbound, processing, and processed. This folder structure gives you a visual state machine. Files move left to right as the pipeline handles them, and anyone with workspace access can see where a document stands at a glance.

Next, generate an API key. Navigate to your organization settings and create a key scoped to the ingestion workspace. Restrict permissions to file read, file write, and metadata operations. Avoid granting full admin scope to a processing service.

import httpx

FASTIO_API_KEY = "your-api-key"
BASE_URL = "https://api.fast.io/current"

headers = {
    "Authorization": f"Bearer {FASTIO_API_KEY}",
    "Content-Type": "application/json",
}

# Create the inbound folder
response = httpx.post(
    f"{BASE_URL}/storage/folder/",
    headers=headers,
    json={
        "workspace_id": "ws_abc123",
        "name": "inbound",
    },
)
inbound_folder_id = response.json()["id"]

Enable Intelligence Mode on the workspace. This tells Fastio to automatically index every uploaded file for semantic search and RAG. With Intelligence Mode active, documents become queryable the moment they finish uploading. You can verify the indexing status by checking the ai_state field on any file: it progresses from pending to in_progress to ready.

Intelligence Mode consumes credits for every document page it ingests. Plans come with a monthly credit allowance (300,000 on Starter, 1.2 million on Business, 4.5 million on Growth), and storage and seats come with the plan rather than the meter. Watch the allowance if you are indexing at volume.

Automate Your Document Workflows

Start building document processing pipelines on Fastio with plan-included storage, monthly credits, and full MCP server access. Begin with a 14-day trial.

Start 14-Day Trial

To react promptly when documents arrive, listen to Fastio's realtime activity feed or connect to the WebSocket events feed. This eliminates latency while avoiding blind delays.

Your application polls the workspace activity feed or receives event notifications containing the file ID, filename, size, MIME type, workspace ID, and the uploading user or agent.

Here is an asynchronous Python worker that polls the activity feed and enqueues new files for processing:

import httpx
import os
import asyncio

BASE_URL = "https://api.fast.io/current"
FASTIO_API_KEY = os.environ.get("FASTIO_API_KEY", "your-api-key")

async def poll_activity_feed(workspace_id: str):
    headers = {"Authorization": f"Bearer {FASTIO_API_KEY}"}
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"{BASE_URL}/workspace/{workspace_id}/activity/",
            headers=headers,
        )
        events = response.json().get("events", [])
        for event in events:
            if event.get("type") == "file.created":
                file_data = event["data"]
                enqueue_processing_job(
                    file_id=file_data["id"],
                    file_name=file_data["name"],
                    mime_type=file_data["mime_type"],
                    workspace_id=file_data["workspace_id"],
                )

Ensure your event consumer processes each event idempotently using the unique event identifier. This prevents duplicate extraction runs if the same event is received more than once.

Process events quickly by pushing tasks onto a background queue. To avoid bottlenecks, handle the extraction asynchronously in dedicated workers.

The extraction stage is where the pipeline generates value. Your async worker picks up a job from the queue, connects an LLM to the Fastio MCP server, and instructs it to read the document and return structured fields.

Fastio exposes its MCP server via Streamable HTTP at /mcp and legacy SSE at /sse. The server provides a consolidated MCP toolset that let an LLM interact with workspaces, files, metadata, and AI features without your application downloading the raw document. For full tool documentation, see mcp.fast.io/skill.md.

Here is how the extraction flow works in practice. Say your pipeline processes vendor invoices. The worker receives a file ID from the queue and sends a prompt to Claude, GPT-4, or Gemini with access to the Fastio MCP server:

System prompt:
You have access to the Fastio MCP server. Use the AI chat tool
to read the document and extract structured data.

User prompt:
Analyze the file with ID "file_xyz789" in workspace "ws_abc123".
Extract: vendor_name, invoice_number, invoice_date, line_items,
subtotal, tax, and total_due. Return strict JSON.

The LLM calls the Fastio MCP tools to access the file contents. Because Intelligence Mode has already indexed the document, the model can run semantic queries against specific sections rather than loading the entire file into its context window. This matters for long documents. A 40-page contract does not need to fit entirely in the prompt; the agent can search for "payment terms" or "governing law" and read only the relevant passages.

The LLM returns a JSON object with the extracted fields:

{
  "vendor_name": "Acme Supplies Co.",
  "invoice_number": "INV-2026-0847",
  "invoice_date": "2026-03-14",
  "line_items": [
    {"description": "Widget A", "qty": 100, "unit_price": 12.50},
    {"description": "Widget B", "qty": 50, "unit_price": 8.75}
  ],
  "subtotal": 1687.50,
  "tax": 135.00,
  "total_due": 1822.50
}

This approach keeps the raw PDF inside the secure Fastio environment. Your application server only handles the structured output, which reduces bandwidth, simplifies security, and avoids managing temporary file storage on your processing nodes.

For documents that need traditional OCR (scanned images with no selectable text), you can route them to a dedicated OCR service like Google Document AI or Tesseract before sending the extracted text to the LLM for field parsing. The hybrid OCR-plus-LLM pattern catches formatting that pure OCR misses while keeping accuracy high.

Once the LLM returns structured data, your worker writes it back to Fastio as custom metadata on the file. This makes every extracted field searchable and filterable inside the workspace.

Fastio's metadata system supports templates with typed fields (string, int, float, bool, datetime, URL, JSON). Create a metadata template for your document type, assign it to the workspace, and then set values on individual files:

# Set extracted metadata on the processed file
metadata_payload = {
    "vendor_name": extracted["vendor_name"],
    "invoice_number": extracted["invoice_number"],
    "invoice_date": extracted["invoice_date"],
    "total_due": extracted["total_due"],
    "status": "extracted",
}

httpx.put(
    f"{BASE_URL}/storage/file/{file_id}/metadata/",
    headers=headers,
    json=metadata_payload,
)

# Move file from inbound to processed
httpx.post(
    f"{BASE_URL}/storage/move/",
    headers=headers,
    json={
        "file_id": file_id,
        "destination_folder_id": processed_folder_id,
    },
)

After updating metadata and moving the file, the pipeline is complete for that document. A human team member opening the workspace sees the file in the processed folder with all extracted fields visible in the sidebar. They can search for "invoices from Acme Supplies over $1,000" and get instant results because the metadata is indexed.

If the extraction flagged anomalies, like a missing signature field on a contract or a suspiciously high invoice total, your worker can tag the file as needs_review and post a comment mentioning a specific reviewer. Fastio's comment system supports mentions, so the reviewer gets notified without leaving the workspace.

This handoff between agent and human is where Fastio's shared workspace model pays off. The agent does the heavy lifting of reading and tagging. The human reviews, approves, or corrects. Both work in the same environment, on the same files, with a full audit trail of who did what.

For pipelines that need human verification, workers can leave anchored comments on document pages or write extraction summaries to collaborative notes in the workspace. Team members can review the extracted values directly alongside the file and leave feedback before downstream systems ingest the structured data.

A pipeline that works on 10 test documents needs additional safeguards before it handles thousands in production.

Retry with backoff. Wrap all Fastio API calls in exponential backoff. Transient network errors and rate limits are normal at scale. If an extraction fails because the LLM returned malformed JSON, retry once with a stricter prompt. If it fails again, move the file to an errors folder and tag it with status: failed so a human can investigate.

Idempotency. Activity events can be received more than once during reconnects. Store each event identifier in a set (Redis or a database column with a unique constraint) and skip duplicates. Without this, a network hiccup could cause the same document to be processed and tagged twice.

Concurrency controls. When multiple workers process files concurrently, rely on Fastio's file version history, granular permissions, and append-only audit log. Check metadata versions or inspect audit records to track updates and prevent accidental overwrites.

# Update metadata with bearer authentication
response = httpx.patch(
    f"{BASE_URL}/storage/file/{file_id}/metadata/",
    headers=headers,
    json={"metadata": extracted_data},
)
response.raise_for_status()

Scheduled audits. Run a scheduled job (every 15 minutes is usually enough) that verifies files in inbound against your processed database records and enqueues any pending items. This catches anything missed during worker redeployments.

Monitor credit usage. Intelligence Mode indexing consumes credits per page, so a batch of long PDFs can draw down a large slice of the monthly allowance in one run. Track your credit usage through the Fastio dashboard and set alerts before you hit the plan limit, since overage bills at $10 per 100,000 credits.

Handle corrupt files gracefully. Password-protected PDFs, zero-byte uploads, and unsupported formats will appear in any real pipeline. Check file size and MIME type before sending to extraction. If a file is unreadable, skip it, tag it with the error reason, and move on.