S3 Operations

<!-- AUTO-GENERATED by export-skills.py — DO NOT EDIT -->
---
name: s3-operations
description: Amazon S3 operations for AI agent workflows including document staging, knowledge base ingestion, and artifact storage. Use when agents need to read, write, or manage files in S3 for RAG pipelines or data processing.
tags: [aws, s3, storage, data]
---

# Amazon S3 Operations for AI Agents

Patterns for using S3 as the document and artifact layer for AI agent workflows.

## When to Use

- Staging documents for Bedrock Knowledge Base ingestion
- Storing agent-generated artifacts (reports, exports, processed files)
- Reading structured/unstructured data for agent analysis
- Managing multi-tenant document isolation
- Pre-signed URL generation for secure file sharing

## Patterns

### 1. Document Staging for Knowledge Base

```python
import boto3
from pathlib import Path

s3 = boto3.client("s3")

def stage_documents_for_kb(
    bucket: str,
    prefix: str,
    local_dir: str,
    metadata: dict | None = None,
):
    """Upload documents to S3 for Bedrock Knowledge Base ingestion."""
    supported_extensions = {
        ".pdf", ".txt", ".html", ".md", ".csv",
        ".doc", ".docx", ".xls", ".xlsx", ".pptx"
    }

    for path in Path(local_dir).rglob("*"):
        if path.suffix.lower() in supported_extensions:
            key = f"{prefix}/{path.relative_to(local_dir)}"
            extra_args = {}
            if metadata:
                extra_args["Metadata"] = metadata

            s3.upload_file(
                str(path), bucket, key,
                ExtraArgs=extra_args
            )
```

### 2. Multi-Tenant Document Isolation

```python
# S3 key pattern for tenant isolation
# s3://faos-documents/{tenant_id}/knowledge-base/{document_type}/{filename}

def get_tenant_prefix(tenant_id: str, doc_type: str = "general") -> str:
    return f"{tenant_id}/knowledge-base/{doc_type}/"

def list_tenant_documents(bucket: str, tenant_id: str) -> list[dict]:
    """List all documents for a specific tenant."""
    paginator = s3.get_paginator("list_objects_v2")
    documents = []
    for page in paginator.paginate(
        Bucket=bucket,
        Prefix=f"{tenant_id}/knowledge-base/"
    ):
        for obj in page.get("Contents", []):
            documents.append({
                "key": obj["Key"],
                "size": obj["Size"],
                "last_modified": obj["LastModified"].isoformat(),
            })
    return documents
```

### 3. Pre-Signed URLs for Secure Sharing

```python
def generate_download_url(bucket: str, key: str, expires_in: int = 3600) -> str:
    """Generate a pre-signed URL for secure document download."""
    return s3.generate_presigned_url(
        "get_object",
        Params={"Bucket": bucket, "Key": key},
        ExpiresIn=expires_in,
    )
```

### 4. Streaming Large Files for Agent Processing

```python
import json

def stream_jsonl_from_s3(bucket: str, key: str):
    """Stream JSONL file line-by-line without loading into memory."""
    response = s3.get_object(Bucket=bucket, Key=key)
    for line in response["Body"].iter_lines():
        yield json.loads(line)
```

## Anti-Patterns

- Loading entire large files into memory -- use streaming for files > 100MB
- Hardcoding bucket names -- use environment variables or SSM Parameter Store
- Using public buckets for agent data -- always use private buckets with IAM policies
- Skipping server-side encryption -- enable SSE-KMS for regulated data

## References

- [Amazon S3 Documentation](https://docs.aws.amazon.com/s3/)
- [AWS S3 MCP Server](https://github.com/awslabs/mcp)
- [S3 Security Best Practices](https://docs.aws.amazon.com/AmazonS3/latest/userguide/security-best-practices.html)

<!-- Source: .faos/custom/skills/cloud/aws/s3-operations/SKILL.md -->