Lambda Just Got a File System. I Put AI Agents on It.

import boto3
import os

s3 = boto3.client("s3")

def lambda_handler(event, context):
    bucket = event["bucket"]
    key = event["key"]

    # 第一件事：下载到 /tmp
    # S3 不给你文件，只给你对象
    local_path = f"/tmp/{key.split('/')[-1]}"
    s3.download_file(bucket, key, local_path)

    # 读文件，处理
    with open(local_path) as f:
        content = f.read()
    result = process(content)

    # 写回去，上传到 S3
    s3.put_object(Bucket=bucket, Key=f"output/{key}", Body=result)

    # 清理 /tmp，否则下次就没空间了
    os.remove(local_path)

import s3fs

fs = s3fs.S3FileSystem()
with fs.open(f"s3://my-bucket/data/file.txt", "r") as f:
    content = f.read()

用户 S3 事件触发 Lambda
         ↓
Lambda 挂载 S3 bucket 到 /mnt/workspace
         ↓
代码直接读写 /mnt/workspace/*
         ↓
S3 Files 处理后台同步到 S3

from pathlib import Path

WORKSPACE = Path("/mnt/workspace")

def lambda_handler(event, context):
    # 直接从挂载点读取，不需要下载
    content = (WORKSPACE / "source" / "app.py").read_text()
    result = process(content)

    # 直接写到挂载点，不需要上传
    (WORKSPACE / "output" / "result.json").write_text(result)

                    ┌─────────────────────────────────────┐
                    │           S3 Bucket                  │
                    │   (持久化存储，系统 of record)        │
                    │   规模：EB 级别，跨 region 复制        │
                    └──────────────┬──────────────────────┘
                                   │
                    ┌──────────────▼──────────────────────┐
                    │     S3 Files FileSystem              │
                    │   (基于 EFS，桥接 S3 和 NFS)          │
                    │                                      │
                    │   - 挂载点：/mnt/workspace            │
                    │   - 自动同步到 S3（分钟级）           │
                    │   - S3 更改即时可见（秒级）           │
                    │   - 亚毫秒读延迟（缓存热数据）         │
                    └──────────────┬──────────────────────┘
                                   │
              ┌────────────────────┼────────────────────┐
              │                    │                    │
   ┌──────────▼──────────┐  ┌──────▼──────┐  ┌────────▼────────┐
   │  Orchestrator        │  │  Agent A    │  │   Agent B       │
   │  (Durable Function)  │  │ (Security)  │  │    (Style)      │
   │                      │  │            │  │                  │
   │  - GitHub API        │  │ - 读取代码  │  │  - 读取代码      │
   │  - Clone repo        │  │ - 安全分析  │  │  - 风格分析     │
   │  - 启动并行 agents   │  │ - 写结果    │  │  - 写结果        │
   │  - Checkpoint 状态   │  │   JSON      │  │   JSON          │
   └──────────────────────┘  └────────────┘  └─────────────────┘

# Agent A 写入中间状态（供 Agent B 读取）
(WORKSPACE / "state" / "agent_a_progress.json").write_text(
    json.dumps({"step": 2, "files_analyzed": 15, "findings": [...]})
)

S3 Bucket (启用了版本控制)
    ↓
IAM Role (S3 Files 用来访问 bucket)
    ↓
S3 Files FileSystem (桥接 bucket 和 NFS)
    ↓
Mount Targets (每个 AZ 一个，网络端点)
    ↓
Access Point (控制 Lambda 的 POSIX 身份)
    ↓
Lambda Function (挂载 access point)

S3FilesRole:
  Type: AWS::IAM::Role
  Properties:
    Path: /service-role/
    AssumeRolePolicyDocument:
      Version: '2012-10-17'
      Statement:
        - Sid: AllowS3FilesAssumeRole
          Effect: Allow
          Principal:
            Service: elasticfilesystem.amazonaws.com
          Action: sts:AssumeRole
          Condition:
            StringEquals:
              aws:SourceAccount: !Ref AWS::AccountId
            ArnLike:
              aws:SourceArn: !Sub 'arn:aws:s3files:${AWS::Region}:${AWS::AccountId}:file-system/*'

S3FilesBucketPolicy:
  Type: AWS::S3::BucketPolicy
  Properties:
    Bucket: !Ref S3FilesBucket
    PolicyDocument:
      Statement:
        - Effect: Allow
          Principal:
            Service: elasticfilesystem.amazonaws.com
          Action:
            - s3:GetObject
            - s3:PutObject
            - s3:DeleteObject
            - s3:ListBucket
          Resource:
            - !Sub "${S3FilesBucket.Arn}/*"
            - !Sub "${S3FilesBucket.Arn}"
          Condition:
            StringEquals:
              aws:ResourceAccount: !Ref AWS::AccountId

S3FilesAccessPoint:
  Type: AWS::S3Files::AccessPoint
  Properties:
    FileSystemId: !GetAtt S3FileSystem.FileSystemId
    PosixUser:
      Uid: '1000'
      Gid: '1000'
    RootDirectory:
      Path: /lambda
      CreationPermissions:
        OwnerUid: '1000'
        OwnerGid: '1000'
        Permissions: '755'

OrchestratorFunction:
  Type: AWS::Serverless::Function
  DependsOn:
    - MountTargetA
    - MountTargetB
  Properties:
    FunctionName: !Sub "${AWS::StackName}-orchestrator"
    Handler: orchestrator.handler
    Runtime: python3.12
    MemorySize: 1024
    Timeout: 900  # 15 分钟，支持长运行的 durable 流程
    Environment:
      Variables:
        WORKSPACE_PATH: /mnt/workspace
    FileSystemConfigs:
      - Arn: !GetAtt S3FilesAccessPoint.AccessPointArn
        LocalMountPath: /mnt/workspace
    VpcConfig:
      SecurityGroupIds:
        - !GetAtt NetworkingStack.Outputs.LambdaSGId
      SubnetIds:
        - !GetAtt NetworkingStack.Outputs.PrivateSubnetAId
        - !GetAtt NetworkingStack.Outputs.PrivateSubnetBId
    Policies:
      - AmazonS3FilesClientReadWriteAccess
      - AWSLambdaVPCAccessExecutionRole

from strands import Agent
from strands.tools import tool
from pathlib import Path
import json

WORKSPACE = Path("/mnt/workspace")

@tool
def read_code_file(relative_path: str) -> str:
    """Read a code file from the shared workspace.

    Args:
        relative_path: Path relative to the repo root,
                       e.g., 'src/handlers/auth.py'
    Returns:
        File contents as string, or error message.
    """
    full_path = WORKSPACE / "repo" / relative_path
    if not full_path.exists():
        return f"Error: File not found: {relative_path}"
    if not full_path.is_file():
        return f"Error: Not a file: {relative_path}"
    return full_path.read_text()

@tool
def list_code_files(extension: str = ".py") -> list[str]:
    """List all code files in the repo with given extension.

    Args:
        extension: File extension to filter by (default: .py)
    Returns:
        List of relative file paths.
    """
    repo_path = WORKSPACE / "repo"
    return [
        str(p.relative_to(repo_path))
        for p in repo_path.rglob(f"*{extension}")
    ]

@tool
def write_analysis_result(analysis_type: str, content: str) -> str:
    """Write analysis results to shared workspace.

    Args:
        analysis_type: Type of analysis (e.g., 'security', 'style')
        content: JSON string of analysis results
    Returns:
        Path to written file.
    """
    result_path = WORKSPACE / "results" / f"{analysis_type}.json"
    result_path.parent.mkdir(parents=True, exist_ok=True)
    result_path.write_text(content)
    return f"Written to {result_path}"

# 安全审查 agent
security_agent = Agent(
    model="anthropic.claude-3-5-sonnet-20241022",
    tools=[read_code_file, list_code_files, write_analysis_result],
    systemPrompt="""You are a security review agent.

    Your job is to analyze code for security vulnerabilities.

    Focus areas:
    - SQL injection risks (string concatenation in queries)
    - XSS vulnerabilities (unescaped user input in output)
    - Secrets hardcoded in code (API keys, passwords, tokens)
    - Insecure deserialization (pickle, eval on user input)
    - Authentication/authorization bypasses
    - Path traversal vulnerabilities

    Use list_code_files to find relevant files, read_code_file to
    inspect them, and write_analysis_result to save your findings.

    Findings should be structured as JSON with severity (HIGH/MEDIUM/LOW),
    file path, line number if known, and description."""
)

# 代码风格审查 agent
style_agent = Agent(
    model="anthropic.claude-3-5-sonnet-20241022",
    tools=[read_code_file, list_code_files, write_analysis_result],
    systemPrompt="""You are a code style review agent.

    Your job is to analyze code for style issues and maintainability problems.

    Focus areas:
    - Code smells (long methods, deep nesting, duplicated code)
    - Best practice violations (no type hints, magic numbers)
    - Error handling missing or incorrect
    - Naming conventions (unclear variable names, inconsistent casing)
    - Documentation missing or inadequate

    Use list_code_files to find relevant files, read_code_file to
    inspect them, and write_analysis_result to save your findings.

    Findings should be structured as JSON with severity (HIGH/MEDIUM/LOW),
    file path, line number if known, and description."""
)

from aws_lambda_powertools import Logger
from amazon.lambda.durable import DeterministicUuid, when_all, when_any
from pathlib import Path
import json

logger = Logger()

@DeterministicUuid
def orchestration_id(event, context):
    return event["repo_url"]

def lambda_handler(event, context):
    orchestrator = CodeReviewOrchestrator()
    return orchestrator.handle(event, context)

class CodeReviewOrchestrator:
    def __init__(self):
        self.workspace = Path("/mnt/workspace")
        self.repo_path = self.workspace / "repo"
        self.results_path = self.workspace / "results"

    def handle(self, event, context):
        repo_url = event["repo_url"]
        execution_id = context.execution_id

        logger.info(f"Starting code review", extra={"repo": repo_url, "execution": execution_id})

        # Step 1: Clone repository to workspace
        clone_result = self.clone_repo(repo_url)
        if not clone_result["success"]:
            return {"status": "error", "message": clone_result["error"]}

        # Step 2: Launch agents in parallel
        agent_tasks = self.launch_agents(execution_id)

        # Step 3: Wait for all agents to complete
        # (Durable functions handles checkpointing during wait)
        agent_results = self.wait_for_agents(agent_tasks)

        # Step 4: Compile results
        summary = self.compile_results(agent_results)

        return {
            "status": "complete",
            "execution_id": execution_id,
            "summary": summary
        }

    def clone_repo(self, repo_url):
        """Clone GitHub repo to workspace."""
        import subprocess
        try:
            self.repo_path.mkdir(parents=True, exist_ok=True)
            subprocess.run(
                ["git", "clone", "--depth", "1", repo_url, str(self.repo_path)],
                check=True,
                capture_output=True
            )
            return {"success": True}
        except Exception as e:
            logger.error(f"Clone failed: {e}")
            return {"success": False, "error": str(e)}

    def launch_agents(self, execution_id):
        """Start security and style agents in parallel."""
        # In practice, this uses Lambda durable functions to
        # launch child executions for each agent
        return [
            {"agent": "security", "task_id": f"{execution_id}-security"},
            {"agent": "style", "task_id": f"{execution_id}-style"}
        ]

    def wait_for_agents(self, agent_tasks):
        """Wait for all agent tasks to complete."""
        results = {}
        for task in agent_tasks:
            # Read result file written by agent
            result_file = self.results_path / f"{task['agent']}.json"
            if result_file.exists():
                results[task['agent']] = json.loads(result_file.read_text())
            else:
                results[task['agent']] = {"status": "pending"}
        return results

    def compile_results(self, agent_results):
        """Compile individual results into summary."""
        total_findings = sum(
            len(r.get("findings", []))
            for r in agent_results.values()
        )
        return {
            "agents_run": len(agent_results),
            "total_findings": total_findings,
            "by_agent": {
                agent: len(r.get("findings", []))
                for agent, r in agent_results.items()
            }
        }

def lambda_handler(event, context):
    workspace = Path("/mnt/workspace")
    input_dir = workspace / "data" / "raw"
    output_dir = workspace / "data" / "processed"

    # 确保输出目录存在
    output_dir.mkdir(parents=True, exist_ok=True)

    # 处理每个 CSV 文件
    processed_count = 0
    for file_path in input_dir.glob("*.csv"):
        try:
            df = pd.read_csv(file_path)

            # 数据清洗
            cleaned = clean_data(df)

            # 转换
            transformed = transform_data(cleaned)

            # 写出结果（直接写，自动同步到 S3）
            output_path = output_dir / file_path.name
            transformed.to_csv(output_path, index=False)

            processed_count += 1
            logger.info(f"Processed {file_path.name}")

        except Exception as e:
            logger.error(f"Failed to process {file_path}: {e}")

    return {
        "status": "complete",
        "processed": processed_count
    }

# Agent A: 分析第一部分代码，写入进度
(WORKSPACE / "state" / "phase1_findings.json").write_text(
    json.dumps({
        "phase": 1,
        "files_analyzed": 15,
        "findings": [...]
    })
)

# Agent B: 检查 Phase 1 是否完成，决定是否开始
phase1_state = (WORKSPACE / "state" / "phase1_findings.json")
if phase1_state.exists():
    state = json.loads(phase1_state.read_text())
    if state["phase"] == 1:
        # 开始 Phase 2
        pass

from aws_lambda_powertools import Logger
from amazon.lambda.durable import (
    DeterministicUuid,
    orchestration_context,
    current_step
)
import json

logger = Logger()

def lambda_handler(event, context):
    """Entry point for durable orchestration."""
    orchestrator = CodeReviewOrchestrator()
    return orchestrator.handle(event, context)

class CodeReviewOrchestrator:
    def __init__(self):
        self.workspace = Path("/mnt/workspace")

    def handle(self, event, context):
        repo_url = event["repo_url"]

        # 获取当前 orchestrator 状态
        state = orchestration_context()
        current_step_name = current_step()

        if current_step_name == "clone":
            return self.step_clone(repo_url)

        elif current_step_name == "analyze":
            return self.step_analyze()

        elif current_step_name == "summarize":
            return self.step_summarize()

        # 初始调用：从 Step 1 开始
        return self.start_orchestration(repo_url)

    def start_orchestration(self, repo_url):
        """Start the orchestration pipeline."""
        return {
            "steps": [
                {"name": "clone", "action": "clone_repo", "params": {"repo_url": repo_url}},
                {"name": "analyze", "action": "parallel_agents"},
                {"name": "summarize", "action": "compile_results"}
            ]
        }

    def step_clone(self, repo_url):
        """Clone repository (could take a few minutes for large repos)."""
        logger.info(f"Cloning {repo_url}")
        # 这个 step 可能需要几分钟，durable function 会自动 checkpoint
        clone_result = self.clone_repo(repo_url)
        return {
            "next_step": "analyze",
            "checkpoint_data": {"clone_status": "complete"}
        }

    def step_analyze(self):
        """Launch agents and wait for completion."""
        # 在实际实现中，这里会使用
        # durable_function.activity() 来启动 agent 执行
        # 并使用 when_all() 等待它们完成
        return {
            "next_step": "summarize",
            "checkpoint_data": {"agents_status": "complete"}
        }

    def step_summarize(self):
        """Compile and return final results."""
        results = self.read_results()
        return {
            "status": "complete",
            "results": results
        }