How to Implement Cross-Workspace Search with Fastio API

Cross-workspace search with the Fastio API allows centralized systems to securely query data across multiple isolated agent environments using federated search techniques. Most basic search tutorials only cover querying within a single directory or a single repository. However, enterprise developers and AI engineers need multi-tenant search solutions to break down data silos without compromising access controls.

Federated search reduces data duplication across agent memory stores. Instead of copying documents into a massive, centralized database that risks exposing sensitive information across team boundaries, a federated approach keeps the data in its original workspace. The central search application queries each authorized workspace simultaneously, aggregates the results, and presents a unified response to the user or the AI agent. This approach ensures that strict permission boundaries are respected while providing a comprehensive view of organizational knowledge.

Building a global search for multi-agent systems requires managing asynchronous requests, handling pagination, mitigating rate limits, and ranking aggregated results. The Fastio API natively supports the Model Context Protocol (MCP) and offers a consolidated MCP toolset, making it exceptionally straightforward to build these advanced search patterns without managing complex vector databases manually.

• AI Automation Storage API: Persistent Storage for Autonomous AgentsAn AI automation storage API provides programmatic access to persistent file storage that autonomous agents can use to...
• How to Create Fastio Workspaces Programmatically via APIProgrammatic workspace creation allows developers to dynamically spin up isolated, intelligence-ready Fastio...
• How to Build an Agentic Workspace with the Fastio APIAn agentic workspace built with the Fastio API provides persistent memory, file sharing, and tools for autonomous...
• How to Build a Custom RAG Application with Fastio APIBuilding a custom RAG (Retrieval-Augmented Generation) application traditionally requires complex orchestration between...
• Fastio API vs Google Cloud Storage for AI Agents: 2026 ComparisonWhen building persistent memory for AI agents, developers often start with a raw object store. While Google Cloud...
• Fastio API vs Amazon S3: Which is Better for Agent Storage?While Amazon S3 offers raw object storage, Fastio provides an intelligent workspace with built-in MCP tools and...

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

Before you can query multiple Fastio workspaces, you need to establish secure authentication and ensure your agent has the appropriate permissions. Fastio utilizes a robust, token-based authentication mechanism designed for agentic workflows and multi-tenant isolation.

First, acquire your API credentials from the Fastio dashboard. Your agent will need an API key that has read-level access across the target workspaces. It is highly recommended to use the principle of least privilege, assigning only the specific permissions necessary to execute queries rather than full administrative control.

Environment Configuration: Store your credentials securely. In a typical Python application, you might use environment variables:

export FASTIO_API_KEY="your_secure_api_key_here"

Client Initialization: Fastio has no official SDK, so the examples below call the REST API at https://api.fast.io/current/ directly with the requests library and send the API key as a Bearer token.

import os
import requests

API = "https://api.fast.io/current"
HEADERS = {"Authorization": f"Bearer {os.environ['FASTIO_API_KEY']}"}

Ensure that you handle initialization errors robustly. If the key is missing or invalid, your application should fail gracefully rather than attempting unauthorized requests. With the client configured, your application is ready to retrieve the list of accessible workspaces.

To implement cross-workspace search with Fastio API, the first technical step is identifying all the workspaces your agent is authorized to access. You cannot search what you do not know exists. Fastio provides the GET /current/workspaces/all/ endpoint specifically for this purpose.

This endpoint returns an array of workspace objects. Each object contains crucial metadata, including the workspace id (a 19-digit profile ID), the workspace name, and an optional description.

Retrieving the Workspace List:

def get_accessible_workspaces():
    try:
        response = requests.get(f"{API}/workspaces/all/", headers=HEADERS)
        response.raise_for_status()
        workspaces = response.json()["workspaces"]
        print(f"Discovered {len(workspaces)} accessible workspaces.")
        return workspaces
    except Exception as e:
        print(f"Failed to retrieve workspaces: {e}")
        return []

all_workspaces = get_accessible_workspaces()

In a production environment, this list can be dynamic. Workspaces might be created or deleted frequently by other agents or human users. Therefore, it is often best practice to fetch this list dynamically at the time of the search request rather than caching it indefinitely, which risks returning stale data or querying deleted environments. Alternatively, you can subscribe to the events feed over WebSocket or long-poll and refresh the list when membership events arrive, to maintain a synchronized, localized cache of accessible workspace IDs.

Ready to unify your agent workflows?

Get a consolidated MCP toolset and shared workspaces during the trial. Start building intelligent, multi-tenant applications with Fastio today.

Start Business Trial

Once you have the list of workspace IDs, the core of the federated search involves iterating over this list and querying each workspace individually. Fastio features built-in RAG capabilities through its Intelligence Mode, meaning you do not need to build your own vector index. The workspace search endpoint, GET /current/workspace/{workspace_id}/storage/search/, accepts a natural language query and, when Intelligence is enabled, returns ranked semantic matches with text snippets.

The Basic Iteration Pattern:

def run_workspace_search(workspace, search_query, limit=5):
    response = requests.get(
        f"{API}/workspace/{workspace['id']}/storage/search/",
        headers=HEADERS,
        params={"search": search_query, "limit": limit}
    )
    response.raise_for_status()
    files = response.json().get("files") or {}
    
    # Results arrive in ranked order; keep each one's position as its rank
    return [
        {
            "workspace_id": workspace["id"],
            "workspace_name": workspace["name"],
            "node_id": node_id,
            "rank": rank,
            "relevance_score": result.get("relevance_score"),
            "snippet": result.get("content_snippet")
        }
        for rank, (node_id, result) in enumerate(files.items())
    ]

def federated_search(workspaces, search_query, limit_per_workspace=5):
    all_results = []
    
    for workspace in workspaces:
        print(f"Searching in: {workspace['name']} ({workspace['id']})")
        try:
            all_results.extend(run_workspace_search(workspace, search_query, limit_per_workspace))
        except Exception as e:
            print(f"Error querying workspace {workspace['id']}: {e}")
            # Continue to the next workspace even if one fails
            continue
            
    return all_results

This synchronous approach is functional but may become slow if you are querying dozens of workspaces simultaneously. In the next section, we will discuss how to optimize this iteration using asynchronous requests to reduce overall latency.

To build global search for multi-agent systems that responds quickly, you must parallelize your requests. Querying workspaces sequentially creates a bottleneck where the total search time equals the sum of all individual query times. By using asynchronous I/O, you can launch all queries concurrently, reducing the total search time to approximately the duration of the single slowest query.

Concurrent Query Implementation:

With Python's asyncio, you can run each blocking requests call in a worker thread with asyncio.to_thread and wait for all of them together.

import asyncio

async def async_federated_search(workspaces, search_query):
    async def search_single_workspace(workspace):
        try:
            # requests is blocking, so each search runs in a worker thread
            return await asyncio.to_thread(run_workspace_search, workspace, search_query, 5)
        except Exception as e:
            print(f"Query failed for {workspace['id']}: {e}")
            return []
    
    # Create a list of concurrent tasks
    tasks = [search_single_workspace(ws) for ws in workspaces]
    
    # Execute all tasks simultaneously
    nested_results = await asyncio.gather(*tasks)
    
    # Flatten the list of lists
    flat_results = [item for sublist in nested_results for item in sublist]
    return flat_results

When parallelizing, you must be extremely mindful of API constraints. Every response carries rate limit headers (x-ve-limit-avail for requests remaining, x-ve-limit-max for the window cap and x-ve-limit-expires for when the window resets). If your agent queries many workspaces at once and exhausts the window, the API returns HTTP 429 Too Many Requests; back off until x-ve-limit-expires.

To mitigate this, implement a concurrency limiter, such as an asyncio.Semaphore, to restrict the number of inflight requests to a safe threshold.

The final stage of cross-workspace search is taking the unorganized collection of results from various workspaces and presenting them cohesively. The search endpoint returns each workspace's results already ranked, with a relevance_score for each match. That score is normalized within each response (the best match in every workspace scores 1.0), so it cannot be compared across workspaces.

Global Ranking Strategy:

Instead of sorting by score, merge the lists by rank position: every workspace's best match first, then every workspace's second match, and so on.

def rank_global_results(all_results, max_final_results=10):
    # Interleave by each result's position in its own workspace's ranking
    sorted_results = sorted(all_results, key=lambda x: x['rank'])
    
    # Return only the top N results across all workspaces
    return sorted_results[:max_final_results]

This interleaving keeps each workspace's own ordering intact and gives every workspace's best matches a place near the top of the combined list.

Handling Result Display:

When displaying these results to a human user or passing them to an LLM, always preserve the workspace context. A user needs to know not just what the information is, but where it lives. Include the workspace_name and a link or ID to the specific document. This is critical for agents utilizing the Fastio MCP tools, as they require the workspace ID to perform subsequent actions, such as retrieving file metadata, acquiring an advisory file lock before writing, or generating branded shares.

By following these steps: discovery, asynchronous iteration, rate limit management, and global ranking, you can implement a powerful, federated search experience that leverages the full intelligence of the Fastio platform across complex, multi-tenant architectures.