Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.meshagent.com/llms.txt

Use this file to discover all available pages before exploring further.

​
Overview

The StorageClient is the Room API for room-scoped files and folders. Use it to store shared inputs, outputs, artifacts, and attachments that humans, agents, and services in the same room need to read or write.

​
CLI commands

Start with the CLI help, then use a few common commands:
bash
meshagent room storage --help
meshagent room storage ls --room myroom room://
meshagent room storage cp ./report.pdf room://artifacts/report.pdf --room myroom
meshagent room storage show --room myroom room://artifacts/report.pdf

​
Why use the Storage API?

  • Share files between people, agents, and deployed services without wiring separate object storage access.
  • Store outputs such as reports, logs, generated media, or intermediate artifacts.
  • Download file contents directly or get a downloadable URL when you need to hand a file to another system.

​
How it works

Paths are relative to the room storage root. Use upload or uploadStream to write files, download or downloadStream to read them back, and download_url / downloadUrl when you need a fetchable URL for another system. The API also emits file.updated, file.deleted, and file.moved events so other participants can react to changes.

​
Permissions and grants

upload / uploadStream and download / downloadStream / download_url require a storage grant on the participant token for the target path. Storage grants are path-based, which lets you give a service access to only part of room storage. In the current server implementation, list, exists, stat, and delete are not gated by storage grants. Use API Scopes and Service YAML for the full scope model.

​
Events

The storage system emits three types of events:
  • file.updated
    Triggered when a file is created or updated. Python and .NET handlers receive path and participant_id. JS/TS and Dart handlers receive a FileUpdatedEvent with .path.
def on_file_updated(path: str, participant_id: str):
   print(f"File updated: {path} by {participant_id}")

room.storage.on("file.updated", on_file_updated)
  • file.deleted
    Triggered when a file is deleted. Python and .NET handlers receive path and participant_id. JS/TS and Dart handlers receive a FileDeletedEvent with .path.
def on_file_deleted(path: str, participant_id: str):
   print(f"File deleted: {path} by {participant_id}")

room.storage.on("file.deleted", on_file_deleted)
  • file.moved Triggered when a file or folder is moved. Python and .NET handlers receive source_path, destination_path, and participant_id. JS/TS and Dart handlers receive a FileMovedEvent with .sourcePath and .destinationPath.
def on_file_moved(source_path: str, destination_path: str, participant_id: str):
   print(f"File moved: {source_path} -> {destination_path} by {participant_id}")

room.storage.on("file.moved", on_file_moved)
You can remove an event handler with: 
room.storage.off("file.updated", on_file_updated)
room.storage.off("file.deleted", on_file_deleted)
room.storage.off("file.moved", on_file_moved)

​
API reference

Use the methods below to inspect room storage, upload and download files, and clean up stored artifacts. Each method is asynchronous, so you should await the call.

​
exists(path)

Description
Checks if a file or folder exists at the given path. Parameters
  • path (str): The path to check.
Returns
  • (bool): True if the file or folder exists; False otherwise.
Example: 
meshagent room storage exists \
  --room myroom \
  room://folder/data.json

​
stat(path)

Description
Fetch basic metadata (exists, folder flag, created/updated timestamps) for a file or folder. Parameters
  • path (str): The path to inspect.
Returns
  • StorageEntry | None: Entry when present; None if not found.
The server returns name, is_folder, created_at, and updated_at fields. SDKs that expose stat parse timestamps into native datetime types. Example: 
entry = await room.storage.stat(path="folder/data.json")
if entry:
    print(entry.name, entry.is_folder, entry.created_at, entry.updated_at)
  • Availability: Python SDK exposes stat; other SDKs can call the same endpoint via a custom request until a helper is added.

​
upload(path, data, overwrite=False)

Description
Uploads a complete file payload in one call. Use this when you already have the bytes in memory. Parameters
  • path (str): The destination file path.
  • data (bytes / Uint8Array / byte[]): The file contents to write.
  • overwrite (bool, optional): Whether to replace an existing file at that path.
Returns
  • None
Example: 
data_to_write = b"Hello, Storage!"

await room.storage.upload(
    path="files/new.txt",
    data=data_to_write,
    overwrite=True,
)

​
upload_stream(path, chunks, ...)

Description
Streams file contents chunk-by-chunk. Use this when you do not want to materialize the whole file in memory at once. Parameters
  • path (str): The destination file path.
  • chunks: An async iterable / stream / IAsyncEnumerable<byte[]> of file chunks.
  • overwrite (optional): Replace an existing file at that path.
  • size (optional): Total byte size when you already know it.
  • chunk_size / chunkSize (optional): Pull size for streamed uploads.
Returns
  • None
Example: 
async def chunks():
    yield b"Hello, "
    yield b"world!"


await room.storage.upload_stream(
    path="logs/output.txt",
    chunks=chunks(),
    overwrite=True,
    size=13,
)

​
download_stream(path, chunk_size=...)

Description
Streams a file back as metadata plus data chunks. Use this for larger files or when you want to process the response incrementally. Parameters
  • path (str): The file path to download.
  • chunk_size / chunkSize (optional): Requested chunk size for streamed downloads.
Returns
  • A stream / async iterator of binary chunks. The first chunk carries metadata such as file name, MIME type, and size.
Example: 
stream = await room.storage.download_stream(path="files/report.pdf")

async for chunk in stream:
    if chunk.headers["kind"] == "start":
        print(chunk.headers["name"], chunk.headers["size"])
        continue
    print(f"Received {len(chunk.data)} bytes")

​
download(path)

Description
Retrieves the content of a file from the remote storage. This loads the entire file into memory; use download_url for large files or streaming. Parameters
  • path (str): The file path to download.
Returns
  • File-like Response: Contains the file’s raw data, typically accessible through a .data property.
Example: 
meshagent room storage cp \
  room://files/data.bin \
  ./data.bin \
  --room myroom

​
download_url(path)

Description
Requests a downloadable URL for the specified file path, which can be used to fetch the file directly (e.g., via HTTP). The exact protocol or format of the returned URL may vary, and may be a signed URL if the backing storage provider supports it. Parameters
  • path (str): The file path to retrieve a download URL for.
Returns
  • (str): A URL string you can fetch with your own HTTP or other suitable client.
Example: 
url = await room.storage.download_url(path="files/report.pdf")
print("Download the file from:", url)

​
list(path)

Description
Lists the contents of a folder, returning file and subfolder names along with a flag indicating if each entry is a folder. The server also returns created_at and updated_at timestamps; the Python SDK exposes them on StorageEntry. Parameters
  • path (str): The folder path to list.
Returns
  • (list): A list of entries, each containing a name and is_folder property. Timestamps are available in SDKs that surface them.
Example: 
meshagent room storage ls \
  --room myroom \
  room://some_folder

​
delete(path)

Description
Deletes a file or folder at the given path. A file.deleted event is typically emitted afterward. To delete a folder, pass recursive=True (Python) or call the raw storage.delete request with the recursive flag. Parameters
  • path (str): The file path to delete.
  • recursive (bool, optional): Set True to delete folders recursively (Python helper or raw request).
Returns
  • None
Example: 
meshagent room storage rm \
  --room myroom \
  room://folder/old_file.txt

​
Example Workflow

A common use case:
  1. Check if a file exists.
  2. Upload data if it doesn’t exist.
  3. Later, download the file to verify or use the data.
  4. Delete the file when it’s no longer needed, reacting to the file.deleted event.
import asyncio
from meshagent.api import RoomClient


async def main():
    # Run with:
    # meshagent room connect --room=my-room --identity=participant-name -- python3 storage-download.py
    async with RoomClient() as room:
        data_to_write = b"Hello, Storage!"

        if not await room.storage.exists(path="example.txt"):
            await room.storage.upload(
                path="example.txt",
                data=data_to_write,
                overwrite=True,
            )

        response = await room.storage.download(path="example.txt")
        print("Downloaded content:", response.data)

        await room.storage.delete(path="example.txt")


asyncio.run(main())
This sequence demonstrates basic upload, read, and deletion flows within a single session.