AgentCore ランタイムで Amazon Bedrock マネージドエージェント (OpenAI を使用) の使用を開始する

MyManagedAgent/
├── agentcore/
│   ├── agentcore.json
│   └── cdk/
└── app/MyManagedAgent/
    ├── Dockerfile
    ├── lifecycle/server.py
    ├── otel/collector.yaml
    ├── plugins/acr-report/
    ├── policies/
    │   ├── bma-acr-policy.json
    │   ├── bma-session-trust.json
    │   └── bma-session-policy.json
    ├── pyproject.toml
    ├── client.py
    └── README.md

キーファイル:

ネットワークオプションを に追加しますagentcore create。

agentcore create --name MyManagedAgent --framework BedrockManagedAgents \
  --network-mode VPC --vpc-id <vpc-id> --subnets <subnet-ids> \
  --security-groups <security-group-ids>

exec-server が接続するため、VPC には Bedrock Managed Agents エンドポイントへのルートが必要です。イメージビルドは Codex と Python をダウンロードするため、インターネットへのルートも必要です。「VPC の AgentCore を設定する」を参照してください。

同じプロジェクトがインスタンスコンピューティングタイプで実行されます。プロジェクトにキャパシティープロバイダーを追加し、キャパシティープロバイダーでエージェントを追加します。

agentcore create --name MyManagedAgent --no-agent
cd MyManagedAgent
agentcore add capacity-provider --name MyCapacityProvider \
  --subnets <subnet-ids> --security-groups <security-group-ids> \
  --os LINUX_ARM64 --instance-types c7g.large
agentcore add agent --name MyManagedAgent --framework BedrockManagedAgents \
  --capacity-provider MyCapacityProvider

次に、 を続行しますステップ 2: デプロイする。 はキャパシティープロバイダーとそのオペレーターロールagentcore deployも作成します。

プロジェクトにないキャパシティープロバイダーをアタッチするには、その ARN を に渡しますagentcore create --framework BedrockManagedAgents --capacity-provider <capacity-provider-arn>。「ランタイムインスタンスとキャパシティープロバイダー」を参照してください。

セッションロールはランタイム実行ロールではありません。 は を使用して実行ロールagentcore deployを作成しますpolicies/bma-acr-policy.json。クライアントは、セッションの作成時に role_arnフィールドの Bedrock マネージドエージェントにセッションロールを付与します。

新しいセッションを開始し、 を指定しない場合--role-arn、クライアントは次のステップを実行します。

ロールを変更するには、ファイルを変更します。クライアントは、ロールに追加する他のポリシーを削除しません。を付与するときにロールはチェックされず--session-id、 で付与するロールは変更されません--role-arn。

本番環境では、クライアントに 1 つのランタイムと 1 つのゲートウェイのロールを で付与します--role-arn。ロールを作成するには、 の 2 つのファイルを使用しますpolicies/。パーティション、リージョン、アカウント ID を入力し、 runtime/と をランタイム ID とゲートウェイ ID に変更gateway/*します。リージョン内のすべてのプロジェクトが を共有するため、プロジェクト内のファイルを変更しないでくださいBmaSessionRole-<region>。

クライアントを実行する ID には、セッションロールiam:PassRoleに対する Bedrock Managed Agents クライアントのアクセス許可、bedrock-mantle:CallWithBearerToken、および が必要です。クライアントがロールを作成および修復できるようにするには、 iam:PutRolePolicyで iam:GetRole、iam:CreateRole、iam:UpdateAssumeRolePolicyiam:GetRolePolicy、、 も指定しますarn:aws:iam::<account-id>:role/BmaSessionRole-*。に付与した場合--role-arn、クライアントはこれら 5 つのアクセス許可を必要としません。

信頼ポリシー、セッションロールのアクセス許可、およびクライアントアクセス許可については、「Amazon Bedrock ユーザーガイド」の「セキュリティロールと IAM ロール」を参照してください。

クライアントは openaiパッケージのclient.beta.agentsリソースを使用します。Bedrock マネージドエージェントはエージェントループを実行します。

クライアントはセッションを作成するときに、 でこれらの値を設定しますenvironment。

クライアントはセッションロールextra_bodyにも role_arn を設定します。openai パッケージには、このフィールドのパラメータがありません。

provide_token ヘルパーは認証情報を使用して AWS 、リクエストごとに有効期間の短い Amazon Bedrock ベアラートークンを取得します。

プロジェクトにゲートウェイを追加する場合は、 agentcore deployを使用して の出力からその ARN を指定します--gateway。フラグは、新しいセッションにのみ使用します。クライアントは、 の ARN を使用してゲートウェイをmcpツールとして追加しますconnector_id。Bedrock マネージドエージェントは、セッションロールを使用してゲートウェイを呼び出します。ゲートウェイは MCP バージョン をサポートしている必要があります2025-03-26。そうでない場合、エージェントはゲートウェイツールを取得せず、Bedrock マネージドエージェントはエラーを表示しません。

"""Send an input to a Bedrock Managed Agents session that uses this project's ACR."""

import argparse
import json
import time
from pathlib import Path
from typing import Any

import boto3
from aws_bedrock_token_generator import provide_token
from botocore.exceptions import ClientError
from openai import NotFoundError, OpenAI

BMA_MODEL_ID = "openai.gpt-5.6-luna"
# The agent runs its commands in this directory. The lifecycle server creates it.
WORKSPACE_DIRECTORY = "/home/app/workspace"
# Plugins and skills that the agent can use. The image has the acr-report plugin
# at this path. Add the path of an S3 Files mount to use the skills in a bucket.
CAPABILITY_DIRECTORIES = ["/opt/bma/plugins"]
TURN_END = ("completed", "failed", "cancelled")
TOOL_CALLS = ("mcp_call", "function_call", "web_search_call")
POLICIES = Path(__file__).parent / "policies"
SESSION_ROLE = "BmaSessionRole-{region}"
SESSION_POLICY = "BmaSession"


def show(data: dict[str, Any]) -> None:
    """Prints the session ID, the commands, the tool calls, and the answer."""
    kind = data["type"].removeprefix("agent.session.")
    item = data.get("item") or {}
    if kind == "created":
        print(f"Session {data['session']['id']}")
    elif kind == "turn.output_text.delta":
        print(data["delta"], end="", flush=True)
    elif kind == "turn.item.done" and item.get("type") == "command_execution":
        print(f"\n$ {item['command']}\n{item.get('output') or ''}".rstrip())
    elif kind == "turn.item.done" and item.get("type") in TOOL_CALLS:
        print(f"\nTool {item.get('name') or item['type']} {item.get('status')}")
    elif kind == "error" or kind.split(".")[-1] in TURN_END:
        source = data.get("turn") or data.get("environment") or data.get("session")
        print(f"\n{kind} {(source or data).get('error') or ''}".rstrip())


def load_policy(name: str, partition: str, region: str, account: str) -> str:
    """Reads a policy file and puts in the partition, the Region, and the account."""
    text = (POLICIES / name).read_text()
    for key, value in (("Partition", partition), ("Region", region), ("AccountId", account)):
        text = text.replace("${AWS::" + key + "}", value)
    return text


def session_role(runtime_arn: str) -> str:
    """Creates or repairs the session role, and returns its ARN.

    If the trust policy or the policy of the role is not the same as the file in `policies/`,
    the client writes the file to the role.
    """
    _, partition, _, region, account = runtime_arn.split(":")[:5]
    name = SESSION_ROLE.format(region=region)
    trust = load_policy("bma-session-trust.json", partition, region, account)
    policy = load_policy("bma-session-policy.json", partition, region, account)
    iam = boto3.client("iam")
    changed = False
    try:
        role = iam.get_role(RoleName=name)["Role"]
    except iam.exceptions.NoSuchEntityException:
        role = iam.create_role(RoleName=name, AssumeRolePolicyDocument=trust)["Role"]
        print(f"Created the session role {name}")
        changed = True
    else:
        if role["AssumeRolePolicyDocument"] != json.loads(trust):
            iam.update_assume_role_policy(RoleName=name, PolicyDocument=trust)
            print(f"Updated the trust policy of {name}")
            changed = True
    try:
        current = iam.get_role_policy(RoleName=name, PolicyName=SESSION_POLICY)["PolicyDocument"]
    except iam.exceptions.NoSuchEntityException:
        current = None
    if current != json.loads(policy):
        iam.put_role_policy(RoleName=name, PolicyName=SESSION_POLICY, PolicyDocument=policy)
        if not changed:
            print(f"Updated the policy of {name}")
        changed = True
    if changed:
        # IAM needs some seconds before Bedrock Managed Agents can use a new or changed role.
        time.sleep(15)
    return role["Arn"]


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--runtime", required=True, help="The ACR ARN.")
    parser.add_argument(
        "--session-id",
        help="The BMA session ID. If it does not exist, the client creates a session.",
    )
    parser.add_argument(
        "--input",
        default="Use the acr-report skill to save an ACR report in the workspace.",
    )
    parser.add_argument(
        "--gateway",
        help="The Gateway ARN from the output of `agentcore deploy`.",
    )
    parser.add_argument(
        "--role-arn",
        help="The session role that BMA assumes. If you do not give it, the client creates a role.",
    )
    parser.add_argument("--delete", action="store_true", help="Delete the session.")
    parser.add_argument("--raw", action="store_true", help="Print events as JSON.")
    args = parser.parse_args()
    # BMA must run in the Region of the ACR.
    region = args.runtime.split(":")[3]

    with OpenAI(
        api_key=lambda: provide_token(region=region),
        base_url=f"https://bedrock-mantle.{region}.api.aws/openai/v1",
    ) as client:
        sessions = client.beta.agents.sessions
        session_id = args.session_id
        if session_id:
            try:
                session = sessions.retrieve(session_id).model_dump(warnings=False)
            except NotFoundError:
                print(f"Session {session_id} does not exist.")
                session_id = None
            else:
                if session["environment"].get("runtime_arn") != args.runtime:
                    raise ValueError(f"Session {session_id} uses another ACR.")
        if not session_id and not args.role_arn:
            try:
                args.role_arn = session_role(args.runtime)
            except ClientError as error:
                parser.error(
                    f"The client cannot create or check the session role: {error}. "
                    "Get the IAM permissions in README.md, or give --role-arn."
                )
            print(f"Session role {args.role_arn}")

        if session_id:
            # BMA opens the stream only with stream=true, and the SDK does not send it.
            events = sessions.events.stream(session_id, extra_query={"stream": "true"})
            message = {
                "role": "user",
                "content": [{"type": "input_text", "text": args.input}],
            }
            sessions.events.create(
                session_id,
                events=[{"type": "agent.session.input.message", "input": [message]}],
            )
        else:
            agent: dict[str, Any] = {
                "model": BMA_MODEL_ID,
                "instructions": "Use the available tools to complete the task.",
            }
            if args.gateway:
                # Bedrock Managed Agents calls Gateway with the session role from the service side.
                agent["tools"] = [
                    {"type": "mcp", "server_label": "team_tools", "connector_id": args.gateway}
                ]
                args.input += " Then list the tools that the Gateway gives you."
            events = sessions.create(
                agent=agent,
                environment={
                    "type": "aws_bedrock_agentcore",
                    "runtime_arn": args.runtime,
                    "runtime_qualifier": "DEFAULT",
                    "workspace_directory": WORKSPACE_DIRECTORY,
                    "capability_directories": CAPABILITY_DIRECTORIES,
                },
                input=args.input,
                stream=True,
                # The SDK has no role_arn parameter. BMA assumes this role to call the ACR.
                extra_body={"role_arn": args.role_arn},
            )

        with events:
            for event in events:
                data = event.model_dump(mode="json", warnings=False)
                session_id = session_id or (data.get("session") or {}).get("id")
                if args.raw:
                    print(json.dumps(data), flush=True)
                else:
                    show(data)
                if data["type"].removeprefix("agent.session.turn.") in TURN_END:
                    break

        if args.delete:
            sessions.delete(session_id)
            print(f"\nDeleted session {session_id}")


if __name__ == "__main__":
    main()

microVM ランタイムで、exec-server のホームディレクトリをセッションストレージに配置します。次に、ワークスペース、接続状態、アイドル状態の停止後もCODEX_HOME維持されます。

セッションストレージの詳細については、AgentCore Runtime のファイルシステム設定」を参照してください。

S3 Files アクセスポイントをランタイムにマウントします。その後、バケットが変更されるとスキルが変わり、イメージは変わりません。S3 ファイルファイルシステム、アクセスポイント、およびセキュリティグループのルールについては、AgentCore Runtime のファイルシステム設定」を参照してください。

S3 Files マウントには、VPC ネットワークモードでランタイムが必要です。のネットワークオプションを使用してプロジェクトを作成しますVPC にランタイムをデプロイする。

バケットで、各プラグインをアクセスポイントのルートディレクトリの下にある独自のディレクトリに配置します。plugins/acr-report/: <plugin>/.codex-plugin/plugin.jsonおよび と同じレイアウトを使用します<plugin>/skills/<skill>/SKILL.md。

のランタイムのエントリにマウントを追加しますagentcore/agentcore.json。

{
  "filesystemConfigurations": [
    {
      "s3FilesAccessPoint": {
        "accessPointArn": "arn:aws:s3files:us-east-1:111122223333:file-system/fs-0123456789abcdef0/access-point/fsap-0123456789abcdef0",
        "mountPath": "/mnt/skills"
      }
    }
  ]
}

次に、 CAPABILITY_DIRECTORIES を client.pyに設定し["/opt/bma/plugins", "/mnt/skills"]、デプロイします。

ランタイムのエントリenvVarsに、 などの変数を追加します{ "name": "DISABLE_ADOT_OBSERVABILITY", "value": "true" }。

各イメージビルドは、最新の Codex リリースをインストールします。ライフサイクルサーバーには Codex 0.154.0 以降が必要です。リリースを固定するには、 CODEX_NON_INTERACTIVE=1 の CODEX_RELEASE=<version>の横に を追加しますDockerfile。