Comience a utilizar los agentes gestionados por Amazon Bedrock (con OpenAI) en tiempo de ejecución 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

Archivos clave:

Añada las opciones de red aagentcore create:

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

La VPC debe tener una ruta al punto final de Bedrock Managed Agents, ya que el servidor ejecutivo se conecta a él. La compilación de la imagen descarga Codex y Python, por lo que también necesita una ruta a Internet. Consulte Configurar AgentCore para VPC.

El mismo proyecto se ejecuta en el tipo de cómputo Instances. Añada un proveedor de capacidad al proyecto y, a continuación, añada el agente con el proveedor de capacidad:

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

A continuación, continúe conPaso 2: Implementar. agentcore deploytambién crea el proveedor de capacidad y su función de operador.

Para adjuntar un proveedor de capacidad que no esté en el proyecto, indique su ARN a. agentcore create --framework BedrockManagedAgents --capacity-provider <capacity-provider-arn> Consulte las instancias de ejecución y los proveedores de capacidad.

El rol de sesión no es el rol de ejecución en tiempo de ejecución. agentcore deploycrea el rol de ejecución conpolicies/bma-acr-policy.json. El cliente asigna el rol de sesión a Bedrock Managed Agents en el role_arn campo cuando crea una sesión.

Cuando inicias una nueva sesión y no la das--role-arn, el cliente sigue estos pasos:

Para cambiar el rol, cambie los archivos. El cliente no elimina otras políticas que alguien añada al rol. No comprueba el rol cuando lo --session-id asignas y no cambia el rol con el que lo asignas--role-arn.

En el caso de la producción, asigne al cliente un rol con un Runtime y un gateway--role-arn. Para crear la función, utilice los dos archivos que aparecen enpolicies/. Introduce tu partición, región e ID de cuenta, y cámbialos por tu ID de tiempo de ejecución y tu ID de puerta de enlace. runtime/ gateway/* No cambies los archivos del proyecto, ya que todos los proyectos de la región los compartenBmaSessionRole-<region>.

La identidad que ejecuta el cliente necesita los permisos de cliente de Bedrock Managed Agents y iam:PassRole el rol de sesión. bedrock-mantle:CallWithBearerToken Para permitir que el cliente cree y repare el rol, también asigneiam:GetRole,iam:CreateRole, iam:UpdateAssumeRolePolicyiam:GetRolePolicy, y así iam:PutRolePolicy sucesivamentearn:aws:iam::<account-id>:role/BmaSessionRole-*. Si lo concede--role-arn, el cliente no necesitará estos cinco permisos.

Para conocer la política de confianza, los permisos del rol de sesión y los permisos del cliente, consulte los roles de seguridad e IAM en la Guía del usuario de Amazon Bedrock.

El cliente usa los client.beta.agents recursos del openai paquete. Bedrock Managed Agents ejecuta el ciclo de agentes.

Cuando el cliente crea la sesión, establece estos valores enenvironment:

El cliente también asume role_arn extra_body el rol de sesión. El openai paquete no tiene ningún parámetro para este campo.

El provide_token ayudante usa tus AWS credenciales para obtener un token portador de Amazon Bedrock de corta duración para cada solicitud.

Si agrega una puerta de enlace al proyecto, indique su ARN a partir de la salida de agentcore deploy with. --gateway Usa la marca solo para una sesión nueva. El cliente agrega la puerta de enlace como una mcp herramienta con el ARN incorporado. connector_id Bedrock Managed Agents llama a la puerta de enlace con la función de sesión. La puerta de enlace debe ser compatible con la versión MCP. 2025-03-26 Si no es así, el agente no recibirá ninguna herramienta de puerta de enlace y Bedrock Managed Agents no mostrará ningún 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"
# 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()

En un entorno de ejecución de microVM, coloque el directorio principal del servidor ejecutivo en el almacenamiento de la sesión. Luego, el espacio de trabajo, el estado de la conexión y la CODEX_HOME estancia después de una inactividad se detienen.

Para obtener más información sobre el almacenamiento de sesiones, consulte Configuraciones del sistema de archivos para AgentCore Runtime.

Monte un punto de acceso a S3 Files en el entorno de ejecución. Luego, las habilidades cambian cuando el cubo cambia y la imagen permanece igual. Para conocer las reglas del sistema de archivos S3 Files, el punto de acceso y el grupo de seguridad, consulte Configuraciones del sistema de archivos para AgentCore Runtime.

Un montaje de S3 Files necesita un tiempo de ejecución en modo de red VPC. Cree el proyecto con las opciones de red incluidas. Implemente el tiempo de ejecución en una VPC

En el depósito, coloca cada complemento en su propio directorio, en el directorio raíz del punto de acceso. Usa el mismo diseño queplugins/acr-report/: <plugin>/.codex-plugin/plugin.json y<plugin>/skills/<skill>/SKILL.md.

Añada la montura a la entrada del Runtime enagentcore/agentcore.json:

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

Luego, CAPABILITY_DIRECTORIES client.py configúrelo ["/opt/bma/plugins", "/mnt/skills"] en e impleméntelo.

Añada una variable a envVars en la entrada del tiempo de ejecución, por ejemplo{ "name": "DISABLE_ADOT_OBSERVABILITY", "value": "true" }.

Cada creación de imagen instala la última versión del Codex. El servidor de ciclo de vida necesita Codex 0.154.0 o una versión posterior. Para fijar una versión, añada CODEX_RELEASE=<version> junto aCODEX_NON_INTERACTIVE=1. Dockerfile