开始在 Runtime 上使用亚马逊 Bedrock 托管代理（使用 OpenAI） AgentCore

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>

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。

在生产环境中，为客户端指定一个运行时角色和一个网关角色--role-arn。要扮演角色，请使用中的两个文件policies/。输入您的分区、区域和账户 ID，然后将输入和更改gateway/*为运行时 ID 和网关 ID。 runtime/请勿更改项目中的文件，因为该区域中的所有项目都是共享的BmaSessionRole-<region>。

运行客户端的身份需要 Bedrock Managed Agents 客户端权限和iam:PassRole会话角色。bedrock-mantle:CallWithBearerToken要让客户端创建和修复角色，还需要提供iam:GetRoleiam:CreateRole、iam:UpdateAssumeRolePolicyiam:GetRolePolicy、、iam:PutRolePolicy等arn:aws:iam::<account-id>:role/BmaSessionRole-*。如果您给出--role-arn，则客户端不需要这五个权限。

有关信任策略、会话角色权限和客户端权限，请参阅 A mazon Bedrock 用户指南中的安全和 IAM 角色。

客户端使用openai软件包的client.beta.agents资源。Bedrock 托管代理运行代理循环。

当客户端创建会话时，它会在以下位置设置这些值environment：

客户端还会设置extra_body会role_arn话角色。该openai软件包没有该字段的参数。

provide_token助手使用您的 Amazon 凭证为每个请求获取一个短暂的 Amazon Bedrock 持有者令牌。

如果您向项目添加网关，请从的输出中给出其 ARN。agentcore deploy --gateway仅在新会话中使用该标志。客户端将网关添加为带有 ARN 的mcp工具。connector_idBedrock 托管代理使用会话角色调用网关。网关必须支持 MCP 版本2025-03-26。否则，代理将无法获得网关工具，Bedrock Managed Agents 也不会显示错误。

"""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 运行时中，将执行服务器的主目录放在会话存储器上。然后是工作空间、连接状态，并在空闲停止后CODEX_HOME保持不变。

有关会话存储的更多信息，请参阅 AgentCore Runtime 的文件系统配置。

在运行时中挂载 S3 文件接入点。然后，当存储桶发生变化时，技能会发生变化，图像保持不变。有关 S3 Files 文件系统、接入点和安全组规则，请参阅 AgentCore 运行时的文件系统配置。

S3 文件挂载需要在 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_DIRECTORIESclient.py到["/opt/bma/plugins", "/mnt/skills"]并部署。

例如，envVars在运行时条目中添加一个变量{ "name": "DISABLE_ADOT_OBSERVABILITY", "value": "true" }。

每个镜像版本都会安装最新的 Codex 版本。生命周期服务器需要 Codex 0.154.0 或更高版本。要固定发行版，请在CODEX_RELEASE=<version>旁CODEX_NON_INTERACTIVE=1边添加Dockerfile。