Add async Python SDK client (AsyncAFS) over httpx

sdk/python/README.md

@@ -10,6 +10,12 @@ isolated AFS-backed workspace.
 pip install redis-afs
 ```
 
+For the async client, install the optional `async` extra (installs `httpx`):
+
+```bash
+pip install 'redis-afs[async]'
+```
+
 ## Quick Start
 
 ```python
@@ -39,6 +45,51 @@ with afs.fs.mount(workspaces=[{"name": "foobar"}], mode="rw") as fs:
     fs.write_file("/README.md", "hello")
 ```
 
+## Async usage
+
+The async client mirrors the sync surface, but every network call is a
+coroutine. Install the `async` extra (`pip install 'redis-afs[async]'`) and
+import `AsyncAFS`.
+
+```python
+import os
+from redis_afs import AsyncAFS
+
+async def main():
+    async with AsyncAFS(api_key=os.environ["AFS_API_KEY"]) as afs:
+        workspace = await afs.workspace.create(name="foobar")
+        async with await afs.fs.mount(
+            workspaces=[{"name": workspace["name"]}],
+            mode="rw",
+        ) as fs:
+            await fs.write_file("/src/README.md", "hello world")
+            result = await fs.bash().exec("cat /foobar/src/README.md")
+            print(result.stdout)
+```
+
+Note that `afs.fs.mount(...)` is itself a coroutine, so the mounted filesystem
+is entered with `async with await afs.fs.mount(...)`.
+
+The async client matters most inside an async server. A blocking HTTP call in
+an `async` handler stalls the event loop and blocks every other request; the
+async client `await`s its network I/O instead, so the loop stays free.
+
+```python
+from fastapi import FastAPI
+from redis_afs import AsyncAFS
+
+app = FastAPI()
+
+@app.get("/readme/{workspace}")
+async def read_readme(workspace: str):
+    async with AsyncAFS() as afs:
+        async with await afs.fs.mount(workspaces=[{"name": workspace}], mode="ro") as fs:
+            return {"content": await fs.read_file(f"/{workspace}/README.md")}
+```
+
+Prefer `async with` (or call `await afs.aclose()` and `await fs.aclose()`) so
+the underlying `httpx` connection pools are closed.
+
 ## Authentication
 
 ```bash

sdk/python/api-docs.md

@@ -52,6 +52,71 @@ fs.bash().exec(command, ...)    # run a shell command against the mount
 | Self-managed endpoint | `AFS_API_BASE_URL` or `AFS(base_url=...)` |
 | Available since | `redis-afs` `0.1.0` |
 
+## Async API
+
+The async client mirrors the sync surface as coroutines, backed by an
+`httpx.AsyncClient`. It requires the optional `async` extra
+(`pip install 'redis-afs[async]'`). Importing `redis_afs` works without
+`httpx`, but constructing an async client without it raises `AFSError` with an
+install hint.
+
+### Syntax
+
+```python
+from redis_afs import AsyncAFS
+
+afs = AsyncAFS(
+    api_key=None,
+    base_url=None,
+    timeout=30.0,
+    headers=None,
+)
+```
+
+`AsyncAFS` takes the same parameters as `AFS`, with the same `AFS_API_KEY` and
+`AFS_API_BASE_URL` environment fallbacks.
+
+### API Methods
+
+```python
+await afs.workspace.create(...)       # create a workspace
+await afs.workspace.list()            # list workspaces
+await afs.workspace.get(workspace)    # get one workspace
+await afs.workspace.fork(...)         # fork a workspace
+await afs.workspace.delete(workspace) # delete a workspace
+
+await afs.checkpoint.list(workspace)  # list checkpoints
+await afs.checkpoint.create(...)      # create a checkpoint
+await afs.checkpoint.restore(...)     # restore a checkpoint
+
+fs = await afs.fs.mount(...)          # create an isolated SDK mount (async)
+await fs.read_file(path)              # read a text file
+await fs.write_file(path, content)    # write a text file
+await fs.list_files(path, depth)      # list a directory
+await fs.glob(pattern, ...)           # match paths
+await fs.grep(pattern, **options)     # search file contents
+await fs.checkpoint(name)             # checkpoint mounted workspaces
+await fs.sync_from_remote()           # materialize workspaces locally
+await fs.sync_to_remote()             # write modified local files back
+await fs.bash().exec(command, ...)    # run a shell command against the mount
+```
+
+`afs.fs.mount(...)` is a coroutine, so the mounted filesystem is entered with
+`async with await afs.fs.mount(...)`. Both `AsyncAFS` and `AsyncMountedFS` are
+async context managers and expose `await afs.aclose()` / `await fs.aclose()`;
+prefer `async with` (or call `aclose()`) so the underlying `httpx` connection
+pools are closed. `await afs.fs.mount(..., concurrency=16)` bounds the number of
+in-flight network requests during file sync (default 16).
+
+### Async At A Glance
+
+| Field | Value |
+| --- | --- |
+| Import | `from redis_afs import AsyncAFS` |
+| Requires | `redis-afs[async]` (installs `httpx`) |
+| Transport | `httpx.AsyncClient` (MCP over HTTP) |
+| Available since | `redis-afs` `0.1.0` |
+
 ## Examples
 
 Create a workspace, write a file, run a command, and clean up the temporary
@@ -461,3 +526,17 @@ fs.map_absolute_repo_paths(command)
   currently propagate local file deletion.
 - `MountedFS` does not yet expose `mkdir`, `rename`, `delete`, `stat`, streams,
   or binary file helpers.
+
+### Async Client
+
+- `AsyncMountedFS` bounds *in-flight network requests* during file sync with
+  the `concurrency` kwarg (default 16), but the number of pending coroutine
+  objects is not bounded, so memory scales with tree size. This is acceptable
+  for typical workspaces; a worker-pool design is a possible future
+  improvement.
+- A failed `sync_from_remote()` (for example a mid-sync error) can leave a
+  partially-materialized local root, the same behavior as the sync client.
+- Local disk I/O during sync stays synchronous; only network calls are async.
+  Using `aiofiles` is a possible future enhancement.
+- `bash().exec(...)` raises `asyncio.TimeoutError` on timeout, whereas the sync
+  client raises `subprocess.TimeoutExpired`.

sdk/python/pyproject.toml

@@ -32,6 +32,9 @@ classifiers = [
 ]
 dependencies = []
 
+[project.optional-dependencies]
+async = ["httpx>=0.27"]
+
 [project.urls]
 Homepage = "https://github.com/redis/agent-filesystem/tree/main/sdk/python"
 Repository = "https://github.com/redis/agent-filesystem"

sdk/python/src/redis_afs/__init__.py

@@ -5,8 +5,19 @@
     AFS,
     AFSError,
     MountedFS,
+    MountMode,
     WorkspaceClient,
 )
+from .aio import (
+    AsyncAFS,
+    AsyncBashRunner,
+    AsyncCheckpointClient,
+    AsyncFSClient,
+    AsyncMCPHttpClient,
+    AsyncMountedFS,
+    AsyncRepoClient,
+    AsyncWorkspaceClient,
+)
 
 __all__ = [
     "BashResult",
@@ -15,5 +26,14 @@
     "AFS",
     "AFSError",
     "MountedFS",
+    "MountMode",
     "WorkspaceClient",
+    "AsyncAFS",
+    "AsyncBashRunner",
+    "AsyncCheckpointClient",
+    "AsyncFSClient",
+    "AsyncMCPHttpClient",
+    "AsyncMountedFS",
+    "AsyncRepoClient",
+    "AsyncWorkspaceClient",
 ]

sdk/python/src/redis_afs/_mcp.py

@@ -0,0 +1,47 @@
+"""MCP-over-HTTP wire helpers (endpoint, JSON-RPC framing, result unwrap)."""
+
+from __future__ import annotations
+
+import json
+from typing import Any, Mapping
+
+from .errors import AFSError
+
+
+def build_rpc_body(rpc_id: int, method: str, params: Mapping[str, Any] | None) -> bytes:
+    return json.dumps(
+        {
+            "jsonrpc": "2.0",
+            "id": rpc_id,
+            "method": method,
+            "params": dict(params or {}),
+        }
+    ).encode("utf-8")
+
+
+def parse_rpc_payload(text: str) -> Any:
+    payload = json.loads(text or "{}")
+    if payload.get("error"):
+        error = payload["error"]
+        raise AFSError(str(error.get("message", "MCP request failed")), code=error.get("code"), payload=payload)
+    return payload.get("result")
+
+
+def unwrap_tool_result(result: Any, name: str) -> Any:
+    if isinstance(result, dict) and result.get("isError"):
+        content = "\n".join(item.get("text", "") for item in result.get("content", []))
+        raise AFSError(content or f"MCP tool {name} failed", payload=result)
+    if isinstance(result, dict):
+        return result.get("structuredContent", result)
+    return result
+
+
+def normalize_mcp_endpoint(base_url: str) -> str:
+    trimmed = base_url.strip().rstrip("/")
+    if not trimmed:
+        raise AFSError("base_url is required")
+    return trimmed if trimmed.endswith("/mcp") else f"{trimmed}/mcp"
+
+
+def strip_none(values: dict[str, Any]) -> dict[str, Any]:
+    return {key: value for key, value in values.items() if value is not None}

sdk/python/src/redis_afs/_paths.py

@@ -0,0 +1,60 @@
+"""Remote path normalization and workspace path resolution."""
+
+from __future__ import annotations
+
+import posixpath
+import re
+from pathlib import Path
+from typing import Sequence
+
+from .errors import AFSError
+
+
+def normalize_remote_path(path: str) -> str:
+    raw = path.strip()
+    if not raw:
+        return "/"
+    parts = [part for part in raw.split("/") if part]
+    if ".." in parts:
+        raise AFSError(f"path {path} must not contain '..'")
+    normalized = posixpath.normpath(raw if raw.startswith("/") else f"/{raw}")
+    return "/" if normalized == "." else normalized
+
+
+class MountTable:
+    """Pure path logic for a set of mounted workspaces.
+
+    Owns longest-prefix workspace resolution, the single-workspace fallback,
+    and the absolute-path rewrite used to map remote workspace roots onto a
+    local mirror. Holds no workspace clients and performs no I/O.
+    """
+
+    def __init__(self, workspace_names: Sequence[str]) -> None:
+        self._names = list(workspace_names)
+
+    @property
+    def names(self) -> list[str]:
+        return list(self._names)
+
+    def resolve(self, raw_path: str) -> tuple[str, str]:
+        """Resolve a raw path to ``(workspace_name, remote_path)``."""
+        normalized = normalize_remote_path(raw_path)
+        for name in sorted(self._names, key=len, reverse=True):
+            prefix = f"/{name}"
+            if normalized == prefix:
+                return name, "/"
+            if normalized.startswith(f"{prefix}/"):
+                return name, normalized[len(prefix) :] or "/"
+        if len(self._names) == 1:
+            return self._names[0], normalized
+        choices = ", ".join(f"/{name}" for name in self._names)
+        raise AFSError(f"path {raw_path} must start with one of: {choices}")
+
+    def map_absolute(self, command: str, local_root: str) -> str:
+        """Rewrite absolute ``/workspace`` prefixes to their local mirror paths."""
+        out = command
+        for name in sorted(self._names, key=len, reverse=True):
+            remote_prefix = f"/{name}"
+            local_prefix = str(Path(local_root, name)).replace("\\", "/")
+            out = re.sub(rf"{re.escape(remote_prefix)}(?=/|\s|$)", local_prefix, out)
+        return out

sdk/python/src/redis_afs/aio/__init__.py

@@ -0,0 +1,22 @@
+"""Async public surface for redis_afs, re-exported from the aio subpackage modules."""
+
+from ._http import AsyncMCPHttpClient, httpx
+from ._mount import AsyncBashRunner, AsyncMountedFS, _AsyncMountedWorkspace
+from ._resources import (
+    AsyncAFS,
+    AsyncCheckpointClient,
+    AsyncFSClient,
+    AsyncRepoClient,
+    AsyncWorkspaceClient,
+)
+
+__all__ = [
+    "AsyncMCPHttpClient",
+    "AsyncAFS",
+    "AsyncWorkspaceClient",
+    "AsyncRepoClient",
+    "AsyncCheckpointClient",
+    "AsyncFSClient",
+    "AsyncMountedFS",
+    "AsyncBashRunner",
+]