Inizia a usare Amazon Bedrock Managed Agents (con OpenAI) su Runtime 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

File chiave:

Aggiungi le opzioni di rete a: agentcore create

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

Il VPC deve disporre di un percorso verso l'endpoint Bedrock Managed Agents, poiché il server exec-server si connette ad esso. La build dell'immagine scarica Codex e Python, quindi necessita anche di un percorso verso Internet. Vedi Configure AgentCore for VPC.

Lo stesso progetto viene eseguito sul tipo di calcolo Instances. Aggiungi un provider di capacità al progetto, quindi aggiungi l'agente con il provider di capacità:

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

Quindi continua conFase 2: Implementazione. agentcore deploycrea anche il fornitore di capacità e il suo ruolo di operatore.

Per collegare un provider di capacità che non fa parte del progetto, assegna il relativo ARN a. agentcore create --framework BedrockManagedAgents --capacity-provider <capacity-provider-arn> Vedi Runtime Instances e provider di capacità.

Il ruolo di sessione non è il ruolo di esecuzione di Runtime. agentcore deploycrea il ruolo di esecuzione conpolicies/bma-acr-policy.json. Il client assegna il ruolo di sessione ai Bedrock Managed Agents sul role_arn campo quando crea una sessione.

Quando si avvia una nuova sessione e non si dà nulla--role-arn, il client esegue questi passaggi:

Per modificare il ruolo, modifica i file. Il client non rimuove altre politiche che qualcuno aggiunge al ruolo. Non controlla il ruolo quando date --session-id e non cambia il ruolo con cui date--role-arn.

Per la produzione, assegnate al client un ruolo per un Runtime e un gateway con--role-arn. Per creare il ruolo, usa i due file inpolicies/. Inserisci la partizione, la regione e l'ID dell'account e modifica i valori in runtime/ e gateway/* in Runtime ID e Gateway ID. Non modificate i file del progetto, perché tutti i progetti della regione sono condivisiBmaSessionRole-<region>.

L'identità che gestisce il client necessita delle autorizzazioni del client Bedrock Managed Agents e del iam:PassRole ruolo della sessione. bedrock-mantle:CallWithBearerToken Per consentire al cliente di creare e riparare il ruolo, fornisci ancheiam:GetRole, iam:CreateRole iam:UpdateAssumeRolePolicyiam:GetRolePolicy, e iam:PutRolePolicy così via. arn:aws:iam::<account-id>:role/BmaSessionRole-* Se dai--role-arn, il cliente non ha bisogno di queste cinque autorizzazioni.

Per la policy di fiducia, le autorizzazioni del ruolo di sessione e le autorizzazioni del client, consulta Security and IAM roles nella Amazon Bedrock User Guide.

Il client utilizza le client.beta.agents risorse del openai pacchetto. Bedrock Managed Agents esegue il loop degli agenti.

Quando il client crea la sessione, imposta questi valori inenvironment:

Il client assume anche role_arn extra_body il ruolo di sessione. Il openai pacchetto non ha un parametro per questo campo.

L'provide_tokenhelper utilizza AWS le tue credenziali per ottenere un token Amazon Bedrock bearer di breve durata per ogni richiesta.

Se aggiungi un gateway al progetto, fornisci il relativo ARN dall'output di agentcore deploy with. --gateway Usa il flag solo per una nuova sessione. Il client aggiunge il gateway come mcp strumento con l'ARN in ingresso. connector_id Bedrock Managed Agents chiama il gateway con il ruolo di sessione. Il gateway deve supportare la versione MCP. 2025-03-26 In caso contrario, l'agente non dispone di strumenti di gateway e Bedrock Managed Agents non mostra alcun errore.

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

In un runtime MicroVM, inserisci la home directory dell'exec-server nell'archivio della sessione. Quindi il workspace, lo stato della connessione e CODEX_HOME rimani dopo un'interruzione di inattività.

Per ulteriori informazioni sull'archiviazione delle sessioni, consulta Configurazioni del file system per AgentCore Runtime.

Monta un punto di accesso S3 Files nel Runtime. Quindi le abilità cambiano quando cambia il bucket e l'immagine rimane la stessa. Per il file system S3 Files, il punto di accesso e le regole del gruppo di sicurezza, vedi Configurazioni del file system for Runtime. AgentCore

Un montaggio di file S3 richiede un Runtime in modalità di rete VPC. Crea il progetto con le opzioni di rete attive. Implementa il Runtime in un VPC

Nel bucket, inserisci ogni plugin nella propria directory sotto la directory principale del punto di accesso. Usa lo stesso layout diplugins/acr-report/: <plugin>/.codex-plugin/plugin.json and<plugin>/skills/<skill>/SKILL.md.

Aggiungi il mount alla voce relativa al Runtime inagentcore/agentcore.json:

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

Quindi CAPABILITY_DIRECTORIES client.py impostate ["/opt/bma/plugins", "/mnt/skills"] e distribuite.

Aggiungi una variabile envVars nella voce relativa al Runtime, ad esempio. { "name": "DISABLE_ADOT_OBSERVABILITY", "value": "true" }

Ogni build di immagine installa l'ultima versione del Codex. Il server del ciclo di vita richiede Codex 0.154.0 o versione successiva. Per bloccare una release, aggiungi CODEX_RELEASE=<version> accanto a in. CODEX_NON_INTERACTIVE=1 Dockerfile