API Reference — Surfaces

Built-in injectable surfaces.

Note

OneDriveSurface requires the onedrive extra: pip install "rampart[onedrive]".

surfaces

Built-in injection surfaces for common platforms.

OneDriveSurface

Python

OneDriveSurface(
    *,
    graph_client,
    drive_id,
    folder_path,
    indexing_delay=DEFAULT_INDEXING_DELAY,
)

Injects payloads into a specific OneDrive location.

The surface is fully configured at construction — drive ID, credentials, and target folder path. The inject() method takes only a payload, keeping the injection signature universal across all surfaces.

Uses the Microsoft Graph API (/drives/{drive_id}/...) for file operations. Requires an authenticated GraphServiceClient from msgraph-sdk.

Parameters:

Name	Type	Description	Default
graph_client	GraphServiceClient	An authenticated GraphServiceClient from msgraph-sdk.	required
drive_id	str	The OneDrive drive identifier.	required
folder_path	str	Target folder path within the drive (e.g., "Documents/test-payloads").	required
indexing_delay	float	Seconds to wait after upload for the agent to see the content. OneDrive indexing is typically fast but depends on the consuming application.	DEFAULT_INDEXING_DELAY

Initialize with Graph client and OneDrive location.

Source code in rampart/surfaces/onedrive.py

def __init__(
    self,
    *,
    graph_client: GraphServiceClient,
    drive_id: str,
    folder_path: str,
    indexing_delay: float = DEFAULT_INDEXING_DELAY,
) -> None:
    """Initialize with Graph client and OneDrive location."""
    self._graph_client = graph_client
    self._drive_id = drive_id
    self._folder_path = folder_path.strip("/")
    self._indexing_delay = indexing_delay

drive_id property

Python

drive_id

The OneDrive drive ID.

folder_path property

Python

folder_path

The target folder path.

indexing_delay property

Python

indexing_delay

Seconds to wait after upload for indexing.

inject

Python

inject(*, payload)

Prepare an injection into the configured OneDrive folder.

Returns an InjectionHandle — enter it as an async context manager to activate the injection, exit to clean up.

Parameters:

Name	Type	Description	Default
payload	Payload	The content to inject.	required

Returns:

Type	Description
_OneDriveInjection	An _OneDriveInjection ready to activate via async with.

Source code in rampart/surfaces/onedrive.py

def inject(self, *, payload: Payload) -> _OneDriveInjection:
    """Prepare an injection into the configured OneDrive folder.

    Returns an InjectionHandle — enter it as an async context manager
    to activate the injection, exit to clean up.

    Args:
        payload: The content to inject.

    Returns:
        An ``_OneDriveInjection`` ready to activate via ``async with``.
    """
    return _OneDriveInjection(surface=self, payload=payload)

upload_async async

Python

upload_async(*, payload)

Upload payload content to OneDrive. Returns the item ID.

Uses the small-file upload endpoint (PUT .../root:/{path}:/content), which supports files up to 4 MiB.

Raises:

Type	Description
ValueError	If a binary payload has no artifact path, or if the payload exceeds the 4 MiB small-upload limit.
InfrastructureError	If Graph returns no DriveItem.

Source code in rampart/surfaces/onedrive.py

async def upload_async(self, *, payload: Payload) -> str:
    """Upload payload content to OneDrive. Returns the item ID.

    Uses the small-file upload endpoint
    (``PUT .../root:/{path}:/content``), which supports files
    up to 4 MiB.

    Raises:
        ValueError: If a binary payload has no artifact path, or
            if the payload exceeds the 4 MiB small-upload limit.
        InfrastructureError: If Graph returns no ``DriveItem``.
    """
    filename = f"{payload.id}{payload.format.extension}"
    upload_path = f"{self.folder_path}/{filename}"

    if payload.format.is_binary:
        if payload.artifact is None:
            msg = (
                f"Binary payload format {payload.format.value} "
                "requires an artifact path."
            )
            raise ValueError(
                msg,
            )

        content = payload.artifact.read_bytes()
    else:
        content = payload.content.encode("utf-8")

    if len(content) > _MAX_SMALL_UPLOAD_BYTES:
        msg = (
            f"Payload {payload.id} is {len(content)} bytes, which "
            "exceeds the 4 MiB small-upload limit. Upload sessions "
            "are not yet implemented."
        )
        raise ValueError(
            msg,
        )

    # Graph path-based addressing: root:/{relative-path}:
    # The trailing colon is required by the API.
    drive_item = (
        await self._graph_client.drives.by_drive_id(self.drive_id)
        .items.by_drive_item_id(f"root:/{upload_path}:")
        .content.put(content)
    )

    if drive_item is None or drive_item.id is None:
        msg = (
            "Graph API returned no DriveItem after upload to "
            f"drive={self.drive_id} path={upload_path}"
        )
        raise InfrastructureError(
            msg,
        )

    item_id = drive_item.id
    logger.info(
        "Uploaded payload %s to OneDrive drive=%s path=%s (item=%s)",
        payload.id,
        self.drive_id,
        upload_path,
        item_id,
    )
    return item_id

delete_async async

Python

delete_async(*, item_id)

Delete a file from OneDrive by item ID.

Source code in rampart/surfaces/onedrive.py

async def delete_async(self, *, item_id: str) -> None:
    """Delete a file from OneDrive by item ID."""
    await (
        self._graph_client.drives.by_drive_id(self.drive_id)
        .items.by_drive_item_id(item_id)
        .delete()
    )
    logger.info(
        "Deleted OneDrive item %s from drive=%s",
        item_id,
        self.drive_id,
    )