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

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

Key files:

Add the network options to agentcore create:

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

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

The same project runs on the Instances compute type. Add a capacity provider to the project, and then add the agent with the capacity provider:

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

Then continue with Step 2: Deploy. agentcore deploy also creates the capacity provider and its operator role.

To attach a capacity provider that is not in the project, give its ARN to agentcore create --framework BedrockManagedAgents --capacity-provider <capacity-provider-arn>. See Runtime Instances and capacity providers.

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 provide_token helper uses your AWS credentials to get a short-lived Amazon Bedrock bearer token for each request.

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

import argparse
import json
from typing import Any

from aws_bedrock_token_generator import provide_token
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")


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 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 URL from the output of `agentcore deploy`.",
    )
    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 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 IAM from the service side.
                agent["tools"] = [
                    {
                        "type": "mcp",
                        "server_label": "team_tools",
                        "required": True,
                        "connection_origin": "service",
                        "transport": {"type": "http", "server_url": args.gateway},
                    }
                ]
                args.input += (
                    " Then use the Gateway's documentation and runbook tools to explain"
                    " how to investigate an MCP connection failure. Cite your sources."
                )
            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,
            )

        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()

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.

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. Create the project 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 { "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.