Get started with Amazon Bedrock Managed Agents (with OpenAI) on AgentCore Runtime

MyManagedAgent/
├── agentcore/
│   ├── agentcore.json
│   ├── aws-targets.json
│   └── cdk/
└── app/agent/
    ├── 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

Key files:

Create the project from the environment-python-bma template as shown above. Then add the following fields to the agent Runtime entry in agentcore/agentcore.json:

{
  "networkMode": "VPC",
  "networkConfig": {
    "vpcId": "vpc-0123456789abcdef0",
    "subnets": ["subnet-0123456789abcdef0"],
    "securityGroups": ["sg-0123456789abcdef0"]
  }
}

Replace the IDs with the IDs of your VPC, subnets, and security groups. A container build in VPC mode needs vpcId.

The subnets must have a route to the Bedrock Managed Agents endpoint, because the exec-server connects out to it. The image build runs in the same subnets and downloads Codex and Python, so the subnets also need a route to the internet. See Configure AgentCore for VPC.

The session role is not the Runtime execution role. agentcore deploy creates the execution role with policies/bma-acr-policy.json. The client gives the session role to Bedrock Managed Agents in the role_arn field when it creates a session.

When you start a new session and do not give --role-arn, the client does these steps:

To change the role, change the files. The client does not remove other policies that someone adds to the role. It does not check the role when you give --session-id, and it does not change a role that you give with --role-arn.

For production, give the client a role for one Runtime and one gateway with --role-arn. To make the role, use the two files in policies/. Put in your partition, Region, and account ID, and change the in runtime/ and gateway/* to your Runtime ID and gateway ID. Do not change the files in the project, because all the projects in the Region share BmaSessionRole-<region>.

The identity that runs the client needs the Bedrock Managed Agents client permissions, bedrock-mantle:CallWithBearerToken, and iam:PassRole on the session role. To let the client create and repair the role, also give iam:GetRole, iam:CreateRole, iam:UpdateAssumeRolePolicy, iam:GetRolePolicy, and iam:PutRolePolicy on arn:aws:iam::<account-id>:role/BmaSessionRole-*. If you give --role-arn, the client does not need these five permissions.

For the trust policy, the permissions of the session role, and the client permissions, see Security and IAM roles in the Amazon Bedrock User Guide.

The client uses the client.beta.agents resources of the openai package. Bedrock Managed Agents runs the agent loop.

When the client creates the session, it sets these values in environment:

The client also sets role_arn in extra_body to the session role. The openai package does not have a parameter for this field.

The provide_token helper uses your AWS credentials to get a short-lived Amazon Bedrock bearer token for each request.

If you add a gateway to the project, give its ARN from the output of agentcore deploy with --gateway. Use the flag only for a new session. The client adds the gateway as an mcp tool with the ARN in connector_id. Bedrock Managed Agents calls the gateway with the session role. The gateway must support MCP version 2025-03-26. If it does not, the agent gets no gateway tools, and Bedrock Managed Agents does not show an error.

"""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"
SESSION_ROLE_ARN = "arn:aws:iam::111122223333:role/your-bma-session-role"
WORKSPACE_DIRECTORY = "/home/app/workspace"
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()

Files in the home directory are removed when the Runtime session stops after the idle timeout. On a microVM Runtime, put the home directory of the exec-server on session storage. Then the workspace, the connection state, and CODEX_HOME stay after an idle stop.

A workspace directory on session storage works only on a Runtime that has session storage. For more information about session storage, see File system configurations for AgentCore Runtime.

Mount an S3 Files access point in the Runtime. Then the skills change when the bucket changes, and the image stays the same. For the S3 Files file system, the access point, and the security group rules, see File system configurations for AgentCore Runtime.

An S3 Files mount needs a Runtime in VPC network mode. Configure the Runtime with the network options in Deploy the Runtime in a VPC.

In the bucket, put each plugin in its own directory under the root directory of the access point. Use the same layout as plugins/acr-report/: <plugin>/.codex-plugin/plugin.json and <plugin>/skills/<skill>/SKILL.md.

Add the mount to the entry for the Runtime in agentcore/agentcore.json:

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

Then set CAPABILITY_DIRECTORIES in client.py to ["/opt/bma/plugins", "/mnt/skills"], and deploy.

Add a variable to envVars in the entry for the Runtime, for example:

{
  "envVars": [
    { "name": "DISABLE_ADOT_OBSERVABILITY", "value": "true" }
  ]
}

Each image build installs the latest Codex release. The lifecycle server needs Codex 0.154.0 or later. To pin a release, add CODEX_RELEASE=<version> next to CODEX_NON_INTERACTIVE=1 in the Dockerfile.