How to Implement Cross-Workspace Search with Fast.io API

Understanding Federated Search in Fast.io

Cross-workspace search with the Fast.io 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 Fast.io API natively supports the Model Context Protocol (MCP) and offers up to multiple discrete tools, making it exceptionally straightforward to build these advanced search patterns without managing complex vector databases manually.

Prerequisites and Authentication Setup

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

First, acquire your API credentials from the Fast.io 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: Initialize your Fast.io client. If you are using the official Python SDK, the setup is minimal.

import os
from fastio import FastIOClient

# Initialize the client using the environment variable
client = FastIOClient(api_key=os.environ.get("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.

Step 1: Discovering Authorized Workspaces

To implement cross-workspace search with Fast.io 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. Fast.io provides the list_workspaces() endpoint specifically for this purpose.

This endpoint returns an array of workspace objects. Each object contains crucial metadata, including the unique WorkspaceID (a UUID), the workspace name, and an optional description.

Retrieving the Workspace List:

def get_accessible_workspaces(client):
    try:
        workspaces = client.list_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(client)

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 implement webhooks to listen for workspace creation and deletion events to maintain a synchronized, localized cache of accessible workspace IDs.

Step 2: Iterating Queries Across Workspaces

Once you have the list of workspace IDs, the core of the federated search involves iterating over this list and querying each workspace individually. Fast.io features built-in RAG capabilities through its Intelligence Mode, meaning you do not need to build your own vector index. The query_documents tool accepts a natural language query and returns semantic matches natively.

The Basic Iteration Pattern:

Provide a code snippet showing how to iterate search queries across a retrieved list of authorized workspace IDs:

def federated_search(client, workspaces, search_query, limit_per_workspace=5):
    all_results = []
    
    for workspace in workspaces:
        print(f"Searching in: {workspace.name} ({workspace.id})")
        try:
            workspace_results = client.query_documents(
                workspace_id=workspace.id,
                query=search_query,
                top_k=limit_per_workspace
            )
            
            for result in workspace_results:
                all_results.append({
                    "workspace_id": workspace.id,
                    "workspace_name": workspace.name,
                    "document_id": result.document_id,
                    "score": result.score,
                    "snippet": result.content_snippet
                })
        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.

Step 3: Optimizing with Asynchronous Requests

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:

If you are using Python's asyncio and an asynchronous Fast.io client (or wrapping the synchronous client in thread pools), your code will look significantly different.

import asyncio

async def async_federated_search(async_client, workspaces, search_query):
    async def search_single_workspace(workspace):
        try:
            results = await async_client.query_documents(
                workspace_id=workspace.id,
                query=search_query,
                top_k=5
            )
            # Tag results with their source workspace
            return [ {**res.dict(), "workspace_id": workspace.id} for res in results ]
        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. According to the Fast.io API Documentation, there is a rate limit of 100 requests per minute for standard operations. If your agent attempts to query multiple workspaces simultaneously, you will trigger rate limiting errors (HTTP multiple Too Many Requests).

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

Step 4: Aggregating and Ranking Results

The final stage of cross-workspace search is taking the unorganized collection of results from various workspaces and presenting them cohesively. The query_documents endpoint returns a relevance score for each match. You can use these scores to rank the combined list.

Global Ranking Strategy:

Because the relevance scores are calculated using the same underlying semantic models across all workspaces, they are generally comparable. You can sort the aggregated list directly by score.

def rank_global_results(all_results, max_final_results=10):
    # Sort the list of dictionaries by the 'score' key in descending order
    sorted_results = sorted(all_results, key=lambda x: x['score'], reverse=True)
    
    # Return only the top N most relevant documents across all workspaces
    return sorted_results[:max_final_results]

This unified ranking ensures that the most relevant documents appear first, regardless of which workspace they originated from.

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 Fast.io MCP tools, as they require the WorkspaceID to perform subsequent actions, such as acquiring a file lock or executing an approval workflow.

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 Fast.io platform across complex, multi-tenant architectures.