AgentCore Code Interpreter × GenU で web 検索結果から PowerPoint を自動生成してみた

import ast
import base64
import json
import os
import uuid

import boto3
from strands import Agent, tool
from bedrock_agentcore.runtime import BedrockAgentCoreApp
from model.load import load_model
from strands_tools.code_interpreter import AgentCoreCodeInterpreter
from strands_tools.code_interpreter.models import ReadFilesAction

from strands_tools.tavily import tavily_search

app = BedrockAgentCoreApp()
log = app.logger

TAVILY_PROVIDER_NAME = "TavilyApiKey"

def _load_tavily_api_key() -> None:
    """Secrets Manager から Tavily API キーを取得し、
    `tavily_search` ツールが参照する環境変数 `TAVILY_API_KEY` に設定する。

    AgentCore Identity の workload identity (per-user OAuth) は使わず、
    Runtime の IAM ロール権限で `bedrock-agentcore-control` 経由で
    シークレット ARN を引き、Secrets Manager から直接値を取得する。
    チーム共有の API キーをエージェント全体で 1 つだけ使うユースケース向け。
    """
    if "TAVILY_API_KEY" in os.environ:
        return

    region = os.environ.get("AWS_REGION", "ap-northeast-1")

    # provider 名から Secrets Manager のシークレット ARN を解決
    control = boto3.client("bedrock-agentcore-control", region_name=region)
    provider = control.get_api_key_credential_provider(name=TAVILY_PROVIDER_NAME)
    secret_arn = provider["apiKeySecretArn"]["secretArn"]

    # シークレットから API キーの値を取得
    secrets = boto3.client("secretsmanager", region_name=region)
    secret_string = secrets.get_secret_value(SecretId=secret_arn)["SecretString"]

    # SecretString は JSON 形式 ({"api_key": "..."} 等) または raw 文字列のことがある
    try:
        parsed = json.loads(secret_string)
        api_key = (
            parsed.get("api_key")
            or parsed.get("apiKey")
            or next(iter(parsed.values()))
        )
    except (json.JSONDecodeError, TypeError):
        api_key = secret_string

    os.environ["TAVILY_API_KEY"] = api_key
    log.info("Loaded TAVILY_API_KEY from Secrets Manager")

# Code Interpreter の初期化
code_interpreter = AgentCoreCodeInterpreter(
    region=os.environ.get("AWS_REGION", "us-east-1")
)

WORKSPACE_DIR = "/tmp/ws"

def _is_under_dir(path: str, base_dir: str) -> bool:
    """パストラバーサル対策: path が base_dir 配下にあるか検証する。"""
    real_path = os.path.realpath(path)
    real_base = os.path.realpath(base_dir)
    return real_path == real_base or real_path.startswith(real_base + os.sep)

@tool
def download_file_from_sandbox(sandbox_filepath: str) -> str:
    """Download a file from the Code Interpreter sandbox to the local workspace.

    The Code Interpreter runs in an isolated sandbox. Files generated there
    are not directly accessible from this runtime. Use this tool to pull
    files from the sandbox before uploading to S3.

    Returns the local file path where the file was saved.

    Args:
        sandbox_filepath: The path to the file in the Code Interpreter sandbox.
                          Use a relative path from the sandbox working directory.
    """
    os.makedirs(WORKSPACE_DIR, exist_ok=True)

    log.info(f"Attempting to download: {sandbox_filepath}")

    result = code_interpreter.read_files(
        ReadFilesAction(type="readFiles", paths=[sandbox_filepath])
    )

    log.info(f"read_files status: {result.get('status')}")

    if result.get("status") != "success":
        raise RuntimeError(f"Failed to read sandbox file: {result}")

    try:
        content_str = result["content"][0]["text"]
    except (KeyError, IndexError, TypeError) as e:
        raise RuntimeError(f"Unexpected read_files response: {result}") from e

    log.info(f"content_str length: {len(content_str)}, first 200 chars: {content_str[:200]}")

    try:
        content_list = json.loads(content_str)
    except (json.JSONDecodeError, TypeError):
        content_list = ast.literal_eval(content_str)

    filename = os.path.basename(sandbox_filepath)
    local_filename = f"{uuid.uuid4()}_{filename}"
    local_path = os.path.join(WORKSPACE_DIR, local_filename)

    for item in content_list:
        if item.get("type") != "resource":
            continue

        resource = item.get("resource", {})
        if "blob" in resource:
            blob_data = resource["blob"]
            log.info(f"blob type: {type(blob_data).__name__}, length: {len(blob_data)}")
            if isinstance(blob_data, bytes):
                # str() → ast.literal_eval() で bytes として復元された場合
                data = blob_data
            else:
                # base64 文字列として返された場合
                data = base64.b64decode(blob_data)
            log.info(f"Decoded data size: {len(data)} bytes")
            with open(local_path, "wb") as f:
                f.write(data)
            log.info(f"Downloaded (binary) {sandbox_filepath} -> {local_path}")
            return local_path
        elif "text" in resource:
            with open(local_path, "w", encoding="utf-8") as f:
                f.write(resource["text"])
            log.info(f"Downloaded (text) {sandbox_filepath} -> {local_path}")
            return local_path

    raise RuntimeError(f"No file content found in read_files response: {content_str[:500]}")

@tool
def upload_file_to_s3_and_retrieve_s3_url(filepath: str) -> str:
    """Upload a file from local workspace to S3 and return the S3 URL.

    GenU's frontend detects S3 URLs and generates its own presigned URL
    on demand, so we return the raw S3 URL instead of a presigned URL.

    Args:
        filepath: The local file path (under /tmp/ws) to upload
    """
    bucket = os.environ.get("FILE_BUCKET")
    if not bucket:
        log.warning("FILE_BUCKET not set")
        return f"S3 upload skipped (FILE_BUCKET not set). Local path: {filepath}"

    if not _is_under_dir(filepath, WORKSPACE_DIR):
        raise ValueError(f"{filepath} is not under {WORKSPACE_DIR}.")

    _, ext = os.path.splitext(filepath)
    key = f"code-interpreter/{uuid.uuid4()}{ext}"

    s3 = boto3.client("s3")
    s3.upload_file(filepath, bucket, key)

    region = s3.meta.region_name
    s3_url = f"https://{bucket}.s3.{region}.amazonaws.com/{key}"
    return s3_url

SYSTEM_PROMPT = """あなたはデータ分析・レポート生成アシスタントです。

## 基本方針
- ユーザーの指示に基づいて、Code Interpreter でデータ分析やファイル生成を行います。
- Excel、CSV、グラフ画像、**PowerPoint (.pptx)**、**PDF** など、さまざまな形式のファイルを生成できます。
- 外部情報が必要なら tavily_search を使います。

## Code Interpreter サンドボックスに渡すコードの制約（厳守）
- `/tmp/` や `/var/` など、`outputs/` 以外のディレクトリにファイルを保存してはならない
- `os.makedirs('/tmp/ws', ...)` のような独自の作業ディレクトリを作成してはならない
- 日本語ファイル名を使用してはならない（英数字とアンダースコアのみ）
- 生成するすべてのファイルは **必ず `outputs/` ディレクトリにのみ** 保存すること
- コードの先頭で必ず `os.makedirs('outputs', exist_ok=True)` を実行すること
- ファイルパスは必ず `outputs/` で始まること（例: `outputs/report.xlsx`）

## PowerPoint (.pptx) 生成の指針
- 生成には `python-pptx` を使用する
- まずユーザーの依頼内容を **スライド構成（タイトル + 各スライドの見出しと要点）** に整理してからコード化する
- 1 スライド = 1 メッセージ。情報を詰め込みすぎず、箇条書きは 5 行以内を目安にする
- 表紙スライドにはタイトル・サブタイトル、本文スライドには見出しと要点、必要に応じて画像や表を入れる
- 日本語テキストを描画する場合は `Pt()` でフォントサイズを明示し、`MSPGothic` など日本語対応フォント名を指定する（指定しないと文字化けは起きないが小さすぎる場合がある）
- web 検索結果から引用する場合は、最後のスライドに **出典 URL の一覧** を載せる

## ファイル提供の手順
Code Interpreter のサンドボックスと Runtime は完全に別のファイルシステムです。
サンドボックスで生成したファイルに Runtime から直接アクセスすることはできません。
ファイルをユーザーに提供するには、以下の 3 ステップを **必ずこの順序で** 実行してください。

1. **Code Interpreter でファイル生成**: `outputs/` ディレクトリに保存する
   os.makedirs('outputs', exist_ok=True)
   df.to_excel('outputs/report.xlsx', index=False)
2. **`download_file_from_sandbox`**: サンドボックスから Runtime にダウンロードする
   - sandbox_filepath には `outputs/ファイル名` の形式で指定すること
   - 戻り値はローカルファイルパス。次のステップにそのまま渡すこと
   download_file_from_sandbox(sandbox_filepath="outputs/report.xlsx")
3. **`upload_file_to_s3_and_retrieve_s3_url`**: ステップ 2 の戻り値を渡して S3 にアップロードする
   - 戻り値はダウンロード URL
   upload_file_to_s3_and_retrieve_s3_url(filepath="<ステップ2の戻り値>")

## URL の出力形式
- 画像: `![ファイル名](URL)`
- Excel/CSV/PDF など: `[ファイル名](URL)`
"""

def process_prompt(prompt):
    """GenU が送る prompt（ContentBlock 配列）からテキストを抽出する。"""
    if isinstance(prompt, str):
        return prompt
    if isinstance(prompt, list):
        texts = [
            block["text"]
            for block in prompt
            if isinstance(block, dict) and "text" in block
        ]
        return "\n".join(texts) if texts else ""
    return str(prompt)

@app.entrypoint
async def invoke(payload):
    log.info("Invoking Code Interpreter Agent...")

    _load_tavily_api_key()

    messages = payload.get("messages", [])
    prompt = process_prompt(payload.get("prompt", ""))

    agent = Agent(
        model=load_model(),
        system_prompt=SYSTEM_PROMPT,
        tools=[
            code_interpreter.code_interpreter,
            download_file_from_sandbox,
            upload_file_to_s3_and_retrieve_s3_url,
            tavily_search
        ],
        messages=messages,
    )

    async for event in agent.stream_async(prompt):
        if "event" in event:
            yield event

if __name__ == "__main__":
    app.run()

import {
  AgentCoreApplication,
  AgentCoreMcp,
  type AgentCoreProjectSpec,
  type AgentCoreMcpSpec,
} from '@aws/agentcore-cdk';
import { CfnOutput, Stack, type StackProps } from 'aws-cdk-lib';
import { Effect, PolicyStatement } from 'aws-cdk-lib/aws-iam';
import { Bucket } from 'aws-cdk-lib/aws-s3';
import { Construct } from 'constructs';

/**
 * GenU の AgentCore スタックが作成した S3 バケット名。
 * FILE_BUCKET 環境変数としてランタイムに渡し、ファイルアップロード先として使用する。
 */
const GENU_FILE_BUCKET_NAME =
  'agentcorestackyakkinhouch-genericagentcoreagentcor-bhatjrupqwqv';

export interface AgentCoreStackProps extends StackProps {
  /**
   * The AgentCore project specification containing agents, memories, and credentials.
   */
  spec: AgentCoreProjectSpec;
  /**
   * The MCP specification containing gateways and servers.
   */
  mcpSpec?: AgentCoreMcpSpec;
  /**
   * Credential provider ARNs from deployed state, keyed by credential name.
   */
  credentials?: Record<string, { credentialProviderArn: string; clientSecretArn?: string }>;
}

/**
 * CDK Stack that deploys AgentCore infrastructure.
 *
 * This is a thin wrapper that instantiates L3 constructs.
 * All resource logic and outputs are contained within the L3 constructs.
 */
export class AgentCoreStack extends Stack {
  /** The AgentCore application containing all agent environments */
  public readonly application: AgentCoreApplication;

  constructor(scope: Construct, id: string, props: AgentCoreStackProps) {
    super(scope, id, props);

    const { spec, mcpSpec, credentials } = props;

    // Create AgentCoreApplication with all agents
    this.application = new AgentCoreApplication(this, 'Application', {
      spec,
    });

    // Create AgentCoreMcp if there are gateways configured
    if (mcpSpec?.agentCoreGateways && mcpSpec.agentCoreGateways.length > 0) {
      new AgentCoreMcp(this, 'Mcp', {
        projectName: spec.name,
        mcpSpec,
        agentCoreApplication: this.application,
        credentials,
        projectTags: spec.tags,
      });
    }

    // GenU の AgentCore スタックが作成した S3 バケットを参照
    const fileBucket = Bucket.fromBucketName(
      this,
      'GenUFileBucket',
      GENU_FILE_BUCKET_NAME
    );

    // エージェント実行ロールに権限を追加
    const agentEnv = this.application.environments.values().next().value!;
    const { role } = agentEnv.runtime;

    // Code Interpreter 権限
    role.addToPrincipalPolicy(
      new PolicyStatement({
        effect: Effect.ALLOW,
        actions: [
          'bedrock-agentcore:CreateCodeInterpreter',
          'bedrock-agentcore:StartCodeInterpreterSession',
          'bedrock-agentcore:InvokeCodeInterpreter',
          'bedrock-agentcore:StopCodeInterpreterSession',
          'bedrock-agentcore:DeleteCodeInterpreter',
          'bedrock-agentcore:ListCodeInterpreters',
          'bedrock-agentcore:GetCodeInterpreter',
          'bedrock-agentcore:GetCodeInterpreterSession',
          'bedrock-agentcore:ListCodeInterpreterSessions',
        ],
        resources: ['*'],
      })
    );

    // Tavily API キーを Secrets Manager から取得するための権限
    // AgentCore Identity の ApiKeyCredentialProvider が作成したシークレットを
    // Runtime ロール権限で直接読み出して環境変数に展開する。
    role.addToPrincipalPolicy(
      new PolicyStatement({
        effect: Effect.ALLOW,
        actions: ['bedrock-agentcore:GetApiKeyCredentialProvider'],
        resources: ['*'],
      })
    );
    role.addToPrincipalPolicy(
      new PolicyStatement({
        effect: Effect.ALLOW,
        actions: ['secretsmanager:GetSecretValue'],
        // AgentCore Identity が作成するシークレットは
        // `bedrock-agentcore-identity!` プレフィックスを持つ
        resources: [
          `arn:aws:secretsmanager:${this.region}:${this.account}:secret:bedrock-agentcore-identity!*`,
        ],
      })
    );

    // GenU バケットへの書き込み権限
    fileBucket.grantWrite(role);

    // 環境変数の設定（GenU のバケット名を渡す）
    agentEnv.runtime.addEnvironmentVariable('FILE_BUCKET', GENU_FILE_BUCKET_NAME);

    // Stack-level output
    new CfnOutput(this, 'StackNameOutput', {
      description: 'Name of the CloudFormation Stack',
      value: this.stackName,
    });
  }
}