Sign in

Python SDK — Storage & File Tools

Use Cosmos Storage/DMS, file adapters, and document tools from the Python SDK

sdkpythonstoragedmsfilestools

Updated 2026-11-03

Verified 2026-11-03

The Python SDK has two file layers:

  • client.storage talks directly to Cosmos Storage / DMS.
  • FileManager uses adapters for local files, DMS-backed files, or MCP-backed file tools.

Use client.storage for normal application upload/download work. Use FileManager and create_document_tools() when an agent should read, create, or edit files through tool callbacks.

Storage Client

Upload bytes to a path:

python
uploaded = await client.storage.upload(
    "/projects/acme/report.txt",
    b"Hello from Cosmos",
    mime_type="text/plain",
    force_create_path=True,
)
print(uploaded["id"])

List and search:

python
items = await client.storage.list_items(path="/projects/acme", recursive=True)
matches = await client.storage.search("report", path="/projects/acme")

Download:

python
content = await client.storage.download("/projects/acme/report.txt")
print(content.decode("utf-8"))

content_by_id = await client.storage.download_item("item-uuid")

url = await client.storage.download_url("/projects/acme/report.txt")
print(url)

Unzip a stored archive:

python
job = await client.storage.unzip_item("archive-item-uuid")
sync_result = await client.storage.unzip_item("archive-item-uuid", sync=True)

Folders and metadata:

python
folder = await client.storage.ensure_folder("/projects/acme/drafts")
item = await client.storage.resolve_path("/projects/acme/report.txt:latest")

if item:
    direct_url = await client.storage.get_download_url(item["id"])

Move and delete:

python
await client.storage.move(item["id"], new_parent_id=folder["id"])
await client.storage.delete(item["id"])

DMS Paths

When resolving a file by path, append :latest if you need the latest version explicitly. Helpers such as download() and download_url() handle this for you.

python
item = await client.storage.resolve_path("/projects/acme/report.txt:latest")

FileManager Adapters

Local workspace:

python
from cosmos_sdk import FileManager, LocalFileSystemAdapter

files = FileManager(LocalFileSystemAdapter("/tmp/workspace"))
await files.write("summary.md", b"# Summary\n")
content = await files.read("summary.md")

Cosmos DMS adapter:

python
from cosmos_sdk import CosmosDMSAdapter, FileManager

files = FileManager(
    CosmosDMSAdapter(
        base_url="https://<storage-host>",
        get_access_token=client.auth.get_access_token,
        org_id="your-org-uuid",
    )
)

MCP-backed files:

python
from cosmos_sdk import FileManager, MCPFileSystemAdapter

files = FileManager(MCPFileSystemAdapter(client.tool_manager))

Document Tools For Agents

create_document_tools() exposes file operations as tools that can be registered with client.tool_manager and used by ADK V2.

python
from cosmos_sdk import FileManager, LocalFileSystemAdapter, create_document_tools

file_manager = FileManager(LocalFileSystemAdapter("/tmp/workspace"))
document_tools = create_document_tools(file_manager)

for name, (definition, handler) in document_tools.items():
    client.tool_manager.register_tool(
        name,
        handler=handler,
        definition=definition,
    )

The tools are read_file, create_file, append_file, and edit_file. Pass an optional on_file_written callback to react to writes. Write tools reject payloads that contain projected artifact-replay content (the placeholder ADK substitutes for large earlier tool output) with TOOL_INVALID_INPUT, so an agent cannot persist an incomplete file by echoing replayed context. Pass complete create, append, or edit payloads.

Keep project-specific file policy outside the SDK. The SDK provides primitives; the application should decide path roots, access checks, retention, and whether agent-created files need review.

See Also