How to Build a Storage Connector for Pydantic AI Agents

Passing mock dictionaries and in-memory text chunks into Pydantic AI dependencies works for basic prototypes, but running agents against real corporate document repositories triggers immediate failure modes: downloading gigabyte-scale files into container memory causes out-of-memory crashes, while unindexed document dumps quickly exhaust LLM context windows and inflate inference costs. Production agent deployments require a dedicated bridge between the reasoning runtime and enterprise file systems.

What Is a Pydantic AI Storage Connector?

A Pydantic AI storage connector is an integration layer that provides autonomous agents with type-safe access to cloud document repositories, enabling structured search and file operations validated against Pydantic models.

In production environments, business intelligence rarely lives inside tidy Python dictionaries or isolated vector databases. It lives across corporate storage drives: sales contracts in Google Drive, quarterly financial models in Microsoft OneDrive, project specifications in Dropbox, and compliance records in Box. A storage connector gives an agent structured access to these repositories without forcing developers to rebuild ingestion pipelines or write brittle ad-hoc API scripts for every workflow.

The Three-Part Production Architecture

A production-grade storage connector decouples document synchronization, background indexing, and agent runtime execution into three distinct layers:

• Cloud storage sync: Connects existing corporate repositories into an organization-owned workspace on a recurring schedule or on demand. The connector keeps enterprise files in their original locations while updating the workspace mirror on a schedule or on demand.
• Remote MCP indexing: Processes incoming documents through workspace intelligence, creating full-text and semantic indices exposed through a standardized Model Context Protocol endpoint over Streamable HTTP.
• Type-validated dependency injection: Injects structured connection state into Pydantic AI agent dependencies, providing agent tools with type-safe query interfaces that return validated Pydantic models rather than raw strings or unstructured blobs.

By establishing this clear separation of concerns, the agent avoids managing low-level storage protocols, handling OAuth refresh tokens, or parsing raw file binaries in memory. Review the architecture patterns in the storage for agents documentation.

The Cost of In-Memory File Processing in Agent Dependencies

Many early Pydantic AI guides recommend passing document loaders or file paths directly into agent dependencies using standard Python data structures. When agents operate against enterprise repositories containing thousands of multi-page files, this approach breaks down across three vectors:

• Container Memory Bloat: Parsing multi-megabyte PDF files, complex presentations, or spreadsheet workbooks inside an agent worker process consumes significant RAM. In containerized cloud environments like Kubernetes or serverless containers, concurrent agent tool calls quickly exceed memory limits and trigger operating system termination.
• Local CPU Latency: Extracting text from complex formats, performing OCR on scanned documents, and computing vector embeddings are CPU-intensive operations. Forcing the agent process to execute these tasks synchronously blocks the agent reasoning loop and inflates response latency.
• Context Window Dilution: Directly parsing large unstructured files within agent dependencies increases token overhead and latency compared to pre-indexed workspace search. Ingesting whole documents into the model context window displaces critical instructions, increases cost per run, and reduces the model's ability to locate relevant facts.

• How to Connect LangGraph to Cloud StorageA LangGraph storage connector bridges cyclical agent graphs with external cloud storage drives, allowing graph nodes to...
• How to Connect Flowise to Cloud StorageA Flowise cloud storage connector links visual LLM workflows to cloud repositories like S3, Google Drive, and Dropbox...
• Google Cloud Storage Connector: Connecting Buckets to AI AgentsA Google Cloud Storage connector links LLM agent frameworks to GCS buckets, providing authenticated endpoints to read,...
• Best Cloud Storage for AI Agents: Top 7 Platforms ComparedCloud storage for AI agents provides persistent file access, version control, and API-driven operations that let...
• How to Give AI Agents Secure Cloud File StorageAutonomous agents need more than vector memory. They need actual file storage to read documents, generate reports, and...
• ChatGPT Cloud Storage: Connect Cloud Drives Without Rate LimitsConnecting ChatGPT to cloud storage often triggers strict rate limits and context saturation when querying multi-file...

More on this subject: Agent Memory and Storage (304 guides)

Cyclic Tool Execution in Pydantic AI

Pydantic AI structures autonomous applications around cyclic tool-calling loops. Unlike basic search pipelines that perform a single retrieval step before generating text, autonomous agents evaluate problems iteratively. An agent examines a customer support inquiry, executes a search tool to find an agreement, inspects the returned section, discovers an ambiguous cross-reference, and triggers a secondary tool call to locate an amendment.

When developers implement tool functions by calling native cloud drive APIs directly, each loop iteration executes multiple network round-trips. The agent must authenticate, list directory contents, filter file metadata, download candidate payloads, and inspect contents sequentially. This repetitive traversal amplifies operational latency and exposes the system to aggressive rate limits.

API Rate Limits and Throttling Across Cloud Drives

Enterprise cloud storage platforms are engineered for human interactive browsing rather than autonomous agent loops. When an agent executes multiple exploratory searches within seconds, cloud storage APIs quickly engage protective throttling mechanisms:

• Microsoft Graph and OneDrive: Microsoft Graph enforces strict per-app and per-tenant concurrency limits. When automated workflows exceed request thresholds, the API returns HTTP 429 Too Many Requests responses, requiring the client to pause and execute exponential backoff.
• Box Developer Rate Limits: Box limits general API requests to 1,000 calls per minute per user. When an iterative agent inspects multiple project folders or submits high-frequency retrieval calls, repetitive requests quickly approach rate limits.
• Google Drive API Quotas: Google Drive enforces per-user per-minute read request quotas. Automated agents performing recursive folder sweeps frequently exceed these thresholds, causing abrupt execution failures.

When an agent encounters rate-limit throttling during a cyclic run, the delay ripples through the entire workflow, causing user-facing timeouts and stranded agent sessions.

Standardized Cloud Connector Benchmark Context

The performance trade-offs between direct cloud drive traversal and pre-indexed workspace retrieval were evaluated in standardized benchmark testing. The connector comparison is published at https://fast.io/benchmarks/ and that page is the only place its numbers live. Fastio was measured the fastest and the lowest cost of the providers tested.

In that evaluation, every measured row is that provider's native connector in Claude Cowork; no row is a local stdio server, a Files-On-Demand stub or SharePoint, and SharePoint was not measured at all. Read the full evaluation report on the Fast.io benchmarks page.

Preserving Existing Corporate Drives with Cloud Sync

Enterprise teams already maintain operational files across established storage systems. Requiring employees to abandon Google Drive, OneDrive, Dropbox, or Box to support AI automation disrupts daily business operations and creates resistance.

The practical architecture preserves existing storage drives while introducing an intelligent retrieval plane. Cloud Sync connects Google Drive, Dropbox, Box and OneDrive folders to a workspace, one-way or two-way, on a schedule or on demand. SharePoint libraries are reached through the OneDrive connector. Synchronization runs periodically or when manually triggered, never continuous, live or real-time.

With one-way sync, corporate cloud storage remains the untouched system of record. Documents added by team members sync into the workspace, where they are indexed for agent access. With two-way sync, agents can write synthesized research briefs, structured extracts, or project deliverables directly into workspace folders, which then sync back to the team's shared cloud drive.

Workspace Intelligence and Hybrid Search

When files sync into a Fast.io workspace with Intelligence Mode enabled, the platform processes incoming documents automatically in the background. Content is parsed across PDFs, Word documents, spreadsheets, presentations, and code files without requiring local agent processing.

The platform provides hybrid search that combines full-text keyword matching, semantic vector search, and search-by-metadata-value. When an agent queries the workspace, it retrieves relevant text chunks accompanied by source file citations, page references, and metadata attributes. This architecture allows the agent to evaluate specific evidence without loading multi-megabyte source files into its context window.

Beyond standard document search, Metadata Views turn unstructured documents into a live, queryable database. By defining extraction fields in natural language, teams extract contract execution dates, counterparty names, policy numbers, or financial line items into structured columns. Agents can query these structured views through the Model Context Protocol, accessing verified tabular data alongside unstructured text.

Workspace Boundaries, Permissions, and Activity Tracking

Enterprise security requires strict boundaries around the documents an agent can reach. Fast.io enforces role-based permissions set per organization, workspace, and share. There are no folder-level or file-level permissions; every member of a workspace reaches the files in that workspace according to their assigned role. When an agent should access only a specific subset of files, administrators assign the agent to a dedicated workspace or a scoped share.

Every workspace maintains per-file version history, so agent modifications never overwrite earlier drafts or destroy team edits. If an agent updates a file or two agents write to the same path, version history retains every iteration.

Workspace operations are recorded in a detailed activity log of changes. The activity log tracks uploads, edits, moves, deletions, permission changes, and share creations across both human users and AI agents. It does not record read-only events such as views, downloads, or previews, providing an auditable record of workspace modifications without accumulating read telemetry noise.

Connect Pydantic AI Agents to Persistent Cloud Storage

Sync your enterprise cloud folders into an intelligent workspace, query pre-indexed documents via remote MCP, and keep agent dependencies lightweight. Monthly plans start with a 14-day free trial.

Start 14-Day Trial

Dependency Injection in Pydantic AI

Pydantic AI approaches agent configuration through explicit type annotations. Rather than relying on global state, monkey-patched functions, or untyped dictionaries, the framework uses a dependency injection pattern that passes runtime state into agent instructions, tools, and output validators.

As noted in the official documentation: Pydantic AI’s dependency system follows established Python practices, making dependencies type-safe, understandable, easy to test, and easy to deploy in production.

When initializing an agent, developers declare a deps_type parameter that defines the structure of runtime dependencies. When the agent runs, the application supplies an instance of that dependency class via the deps argument. Tool functions access this state through a typed RunContext[T] parameter. This design keeps sensitive API credentials and connection clients out of the model prompt while providing tools with direct access to backend services.

Defining Document and Search Schemas

A reliable storage connector defines strict Pydantic schemas for search queries, document metadata, and retrieval results. These schemas validate data entering and leaving the agent, preventing malformed responses from causing runtime exceptions.

from datetime import datetime
from pydantic import BaseModel, Field


class DocumentMetadata(BaseModel):
    file_id: str = Field(description="Unique workspace file identifier")
    name: str = Field(description="Filename including extension")
    path: str = Field(description="Relative path within workspace")
    size_bytes: int = Field(description="File size in bytes")
    updated_at: datetime = Field(description="Timestamp of last modification")
    mime_type: str = Field(description="MIME type of the document")


class SearchFilter(BaseModel):
    file_types: list[str] = Field(
        default_factory=list,
        description="Filter by file extension, such as pdf, docx, or xlsx",
    )
    folders_scope: list[str] = Field(
        default_factory=list,
        description="Restrict search to specific workspace folder paths",
    )


class DocumentChunk(BaseModel):
    chunk_id: str = Field(description="Unique identifier for the text passage")
    file_id: str = Field(description="Identifier of the source file")
    file_name: str = Field(description="Name of the source document")
    content: str = Field(description="Extracted text passage")
    score: float = Field(description="Relevance score between 0.0 and 1.0")
    page_number: int | None = Field(
        default=None, description="Source page number if available"
    )


class SearchResult(BaseModel):
    query: str = Field(description="Original search query text")
    total_matches: int = Field(description="Count of relevant chunks found")
    chunks: list["DocumentChunk"] = Field(
        description="Ordered list of extracted text chunks"
    )

By defining these schemas with Pydantic, the agent framework automatically generates accurate JSON schema representations for the LLM. When the model invokes a search tool, it understands the expected input parameters and receives structured, type-checked results.

Connecting via Remote MCP and Streamable HTTP

The Model Context Protocol establishes an open standard for connecting AI agents to external tools and data sources. Fast.io provides an official hosted MCP server accessible over Streamable HTTP. Setup instructions are detailed in the Fast.io MCP documentation, with tool specifications available in the MCP skill guide.

For custom Python code, agent frameworks, and background scripts, the MCP endpoint is hosted at https://mcp.fast.io/mcp/tools. The server uses Streamable HTTP transport, where every JSON-RPC message is transmitted as an HTTP POST request. Headless applications authenticate by providing a scoped API key via an Authorization: Bearer <your-fastio-api-key> header on the connection.

The /mcp/tools endpoint provides consolidated tools categorized by domain. Document search and read operations run through the storage tool using specific actions, such as search, list, and details. Modifications, deletions, and lock management run through the storage_manage tool. Writes sent to the read tool are refused.

Complete Storage Connector Implementation

The following implementation demonstrates how to build a type-safe storage connector for Pydantic AI using standard HTTP client calls against the remote MCP endpoint. Install the required dependencies:

pip install pydantic-ai httpx

Now implement the storage connector dependency and register the search tool with a Pydantic AI agent:

import os
from dataclasses import dataclass
from typing import Any
import httpx
from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext


class DocumentChunk(BaseModel):
    chunk_id: str = Field(description="Unique identifier for the text passage")
    file_name: str = Field(description="Name of the source document")
    content: str = Field(description="Extracted text passage")
    score: float = Field(description="Relevance score")


class SearchResponse(BaseModel):
    query: str = Field(description="Query executed against workspace")
    results: list["DocumentChunk"] = Field(description="Matching passages")


@dataclass
class FastioStorageConnector:
    api_key: str
    workspace_id: str
    endpoint: str = "https://mcp.fast.io/mcp/tools"
    async def search_documents(
        self, query: str, limit: int = 5
    ) -> list[dict[str, Any]]:
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
            "Accept": "application/json, text/event-stream",
        }
        payload = {
            "jsonrpc": "2.0",
            "id": "search-req-1",
            "method": "tools/call",
            "params": {
                "name": "storage",
                "arguments": {
                    "action": "search",
                    "workspace_id": self.workspace_id,
                    "search": query,
                    "limit": limit,
                },
            },
        }
        async with httpx.AsyncClient(timeout=30.0) as client:
            response = await client.post(
                self.endpoint, headers=headers, json=payload
            )
            response.raise_for_status()
            data = response.json()
            tool_result = data.get("result", {}).get("content", [])
            if tool_result and isinstance(tool_result, list):
                first_block = tool_result[0]
                if (
                    isinstance(first_block, dict)
                    and first_block.get("type") == "text"
                ):
                    import json
                    try:
                        parsed = json.loads(first_block.get("text", "{}"))
                        return parsed.get("matches", [])
                    except json.JSONDecodeError:
                        return []
            return []


@dataclass
class AgentDependencies:
    storage: FastioStorageConnector
    organization_name: str


document_analyst = Agent(
    "openai:gpt-4o",
    deps_type=AgentDependencies,
    system_prompt=(
        "You are an enterprise document research assistant. Use the query_documents "
        "tool to retrieve factual evidence before answering user queries. Always cite "
        "the source document name in your conclusions."
    ),
)


@document_analyst.tool
async def query_documents(
    ctx: RunContext["AgentDependencies"],
    query: str,
    max_results: int = 5,
) -> SearchResponse:
    """Search indexed enterprise workspace documents for relevant passages."""
    raw_matches = await ctx.deps.storage.search_documents(
        query=query, limit=max_results
    )
    chunks = []
    for item in raw_matches:
        chunks.append(
            DocumentChunk(
                chunk_id=str(item.get("id", "unknown")),
                file_name=str(item.get("name", "untitled")),
                content=str(item.get("snippet", "")),
                score=float(item.get("score", 0.0)),
            )
        )
    return SearchResponse(query=query, results=chunks)


async def run_example() -> None:
    api_key = os.environ.get("FASTIO_API_KEY", "your-fastio-api-key")
    workspace_id = os.environ.get("FASTIO_WORKSPACE_ID", "wks_example_123")
    connector = FastioStorageConnector(
        api_key=api_key, workspace_id=workspace_id
    )
    deps = AgentDependencies(
        storage=connector, organization_name="Acme Global"
    )
    result = await document_analyst.run(
        "What are the renewal terms and termination notice periods in our vendor agreements?",
        deps=deps,
    )
    print("Agent Response:")
    print(result.output)

In this implementation, the agent receives only the relevant text snippets necessary to answer the prompt. Source files remain securely stored in the cloud workspace, and the agent worker processes zero unindexed binary payloads.

Advisory File Locks for Concurrent Agents

In production multi-agent environments, multiple specialized agents often interact with the same workspace. A research agent may index files while a summarization agent updates project notes. Without coordination, concurrent writes can create race conditions.

Fast.io provides advisory per-file leases in workspace and share storage. Agents acquire, heartbeat, and release locks through MCP actions: lock-acquire and lock-release on the storage_manage tool, and lock-status on the storage tool.

An advisory file lock operates as a cooperative lease. It does not grant exclusive write rights or prevent all overwrites; if two agents write concurrently, both writes land and version history retains both versions. However, the lease signals to other participating agents and human members that a file is currently being modified. The lease expires automatically unless periodically heartbeated by the holding process, preventing abandoned locks if an agent worker crashes.

Scoping Credentials and Workspace Plans

Securing agent storage integrations requires limiting credentials to the minimum necessary scope. When generating an API key for headless Pydantic AI deployments, administrators scope the credential to specific organizations and workspaces. The agent cannot query or modify files outside its assigned workspace boundary. Consult the agent onboarding guidelines for machine-readable connection details.

Fast.io offers structured subscription tiers designed for teams and agentic workloads. Monthly plans start with a 14-day free trial, which requires a credit card and includes full workspace intelligence capabilities:

Each plan includes dedicated storage capacity, user seats, and monthly AI credit allowances. Credits meter artificial intelligence operations, including hybrid search indexing, semantic retrieval, and Metadata Views extraction. Credit overage, extra storage capacity, and additional bandwidth are billed transparently as documented on the pricing page. Compare plan details on the Fast.io pricing page.

By combining scheduled Cloud Sync from existing cloud drives with remote MCP search in Pydantic AI, engineering teams build high-performance agent workflows that access enterprise document knowledge without the operational penalties of direct drive traversal.