1  {
  2   "cells": [
  3    {
  4     "cell_type": "markdown",
  5     "id": "rkf3pvn7utl",
  6     "source": "# Test AgentCore Deployed Runtime\n\nThis notebook tests the **deployed** AgentCore Runtime via `invoke-agent-runtime`.\n\nIt validates:\n1. Text messages through the deployed endpoint\n2. Image processing (base64 encoded)\n3. Audio transcript processing\n4. Video analysis via TwelveLabs Pegasus\n5. Memory persistence across invocations (short-term + long-term)\n\n> **Prerequisites**: The `AgentAgentCoreStack` must be deployed successfully.",
  7     "metadata": {}
  8    },
  9    {
 10     "cell_type": "code",
 11     "id": "npkagulzftq",
 12     "source": "%pip install boto3 --quiet",
 13     "metadata": {},
 14     "execution_count": null,
 15     "outputs": []
 16    },
 17    {
 18     "cell_type": "code",
 19     "id": "6psoan3fp1",
 20     "source": "import os\nimport botocore.loaders\nimport botocore.session\n\n# Register the custom 'bedrock-agentcore' service model bundled in the deployment package.\n# The system boto3/botocore does not include this service yet.\nCUSTOM_DATA_PATH = os.path.abspath(\n    os.path.join(\n        os.path.dirname(\"__file__\"), \"..\",\n        \"00-agent-agentcore\", \"agent_files\", \"deployment_package\", \"botocore\", \"data\",\n    )\n)\n\ndef agentcore_boto3_session(region: str) -> \"boto3.Session\":\n    \"\"\"Create a boto3 Session that knows about the bedrock-agentcore service.\"\"\"\n    bcore = botocore.session.get_session()\n    loader = botocore.loaders.Loader(\n        extra_search_paths=[CUSTOM_DATA_PATH],\n    )\n    bcore.register_component(\"data_loader\", loader)\n    return boto3.Session(botocore_session=bcore, region_name=region)\n\nprint(f\"Custom botocore data path: {CUSTOM_DATA_PATH}\")\nprint(\"Use agentcore_boto3_session() to create clients for 'bedrock-agentcore'.\")",
 21     "metadata": {},
 22     "execution_count": null,
 23     "outputs": []
 24    },
 25    {
 26     "cell_type": "markdown",
 27     "id": "v2jzjg49g1",
 28     "source": "## 1. Setup - Read stack outputs from SSM",
 29     "metadata": {}
 30    },
 31    {
 32     "cell_type": "code",
 33     "id": "qqn1v5531o",
 34     "source": "import json\nimport boto3\n\nREGION = \"us-east-1\"\n\nssm = boto3.client(\"ssm\", region_name=REGION)\n\ndef get_param(name):\n    return ssm.get_parameter(Name=name)[\"Parameter\"][\"Value\"]\n\nAGENT_RUNTIME_ARN = get_param(\"/agentcore/agent_runtime_arn\")\nS3_BUCKET = get_param(\"/agentcore/s3_bucket_name\")\nMEMORY_ID = get_param(\"/agentcore/memory_id\")\n\n# Extract runtime ID from ARN (last segment after /runtime/)\nAGENT_RUNTIME_ID = AGENT_RUNTIME_ARN.split(\"/\")[-1]\n\nprint(f\"Runtime ARN:  {AGENT_RUNTIME_ARN}\")\nprint(f\"Runtime ID:   {AGENT_RUNTIME_ID}\")\nprint(f\"Memory ID:    {MEMORY_ID}\")\nprint(f\"S3 Bucket:    {S3_BUCKET}\")",
 35     "metadata": {},
 36     "execution_count": null,
 37     "outputs": []
 38    },
 39    {
 40     "cell_type": "markdown",
 41     "id": "dvudxedlt3m",
 42     "source": "## 2. Helper - Invoke Agent Runtime\n\nWraps the `bedrock-agentcore` data plane `invoke-agent-runtime` API.\n\n- `actor_id` identifies the **user** (long-term memory: facts, preferences)\n- `session_id` identifies the **conversation** (short-term memory: turns)",
 43     "metadata": {}
 44    },
 45    {
 46     "cell_type": "code",
 47     "id": "gjuacz71oah",
 48     "source": "session = agentcore_boto3_session(REGION)\nagentcore = session.client(\"bedrock-agentcore\")\n\n# IDs must be >= 33 chars, padded as the agent expects\nTEST_PHONE = \"5730012345670000000000\"\nACTOR_ID  = f\"wa-user-{TEST_PHONE}\".ljust(33, \"0\")   # identifies the USER\nSESSION_ID = f\"wa-chat-{TEST_PHONE}\".ljust(33, \"0\")   # identifies the CONVERSATION\n\nprint(f\"actor_id:   '{ACTOR_ID}' ({len(ACTOR_ID)} chars)\")\nprint(f\"session_id: '{SESSION_ID}' ({len(SESSION_ID)} chars)\")\n\n\ndef invoke_agent(prompt: str, media: dict = None) -> str:\n    \"\"\"Invoke the deployed AgentCore Runtime.\n    \n    Args:\n        prompt: User text message.\n        media: Optional media dict (type, format, data, s3_uri).\n    \n    Returns:\n        Agent response text.\n    \"\"\"\n    payload = {\n        \"prompt\": prompt,\n        \"actor_id\": ACTOR_ID,\n    }\n    if media:\n        payload[\"media\"] = media\n\n    response = agentcore.invoke_agent_runtime(\n        agentRuntimeArn=AGENT_RUNTIME_ARN,\n        runtimeSessionId=SESSION_ID,\n        runtimeUserId=ACTOR_ID,\n        payload=json.dumps(payload).encode(\"utf-8\"),\n    )\n\n    content = []\n    for chunk in response.get(\"response\", []):\n        if isinstance(chunk, bytes):\n            content.append(chunk.decode(\"utf-8\"))\n        elif isinstance(chunk, dict) and \"bytes\" in chunk:\n            content.append(chunk[\"bytes\"].decode(\"utf-8\"))\n\n    response_text = \"\".join(content)\n\n    try:\n        response_json = json.loads(response_text)\n        return response_json.get(\"result\", response_text)\n    except json.JSONDecodeError:\n        return response_text\n\n\nprint(\"\\nHelper ready.\")",
 49     "metadata": {},
 50     "execution_count": null,
 51     "outputs": []
 52    },
 53    {
 54     "cell_type": "markdown",
 55     "id": "qd0mjj0p3u9",
 56     "source": "## 3. Test - Text Message",
 57     "metadata": {}
 58    },
 59    {
 60     "cell_type": "code",
 61     "id": "s47s29w9sr9",
 62     "source": "response = invoke_agent(\"Hola! Mi nombre es Carlos y soy de Colombia. Que puedes hacer por mi?\")\nprint(response)",
 63     "metadata": {},
 64     "execution_count": null,
 65     "outputs": []
 66    },
 67    {
 68     "cell_type": "markdown",
 69     "id": "aufjgi6uzh5",
 70     "source": "## 4. Test - Memory Recall (same session)\n\nThe agent should remember the name and country from the previous turn.",
 71     "metadata": {}
 72    },
 73    {
 74     "cell_type": "code",
 75     "id": "fryovfsb9qv",
 76     "source": "response = invoke_agent(\"Como me llamo y de donde soy?\")\nprint(response)",
 77     "metadata": {},
 78     "execution_count": null,
 79     "outputs": []
 80    },
 81    {
 82     "cell_type": "markdown",
 83     "id": "fvjpvl23lj5",
 84     "source": "## 5. Test - Image Processing\n\nSends a base64-encoded image to the agent via the deployed runtime.",
 85     "metadata": {}
 86    },
 87    {
 88     "cell_type": "code",
 89     "id": "i0ty1gpxsip",
 90     "source": "import base64\nfrom pathlib import Path\n\nIMAGE_PATH = \"imagen2.png\"\n\nif Path(IMAGE_PATH).exists():\n    image_b64 = base64.b64encode(Path(IMAGE_PATH).read_bytes()).decode(\"utf-8\")\n    ext = Path(IMAGE_PATH).suffix.lstrip(\".\")\n    fmt = {\"jpg\": \"jpeg\", \"jpeg\": \"jpeg\", \"png\": \"png\", \"gif\": \"gif\", \"webp\": \"webp\"}.get(ext, \"jpeg\")\n\n    response = invoke_agent(\n        prompt=\"Describe esta imagen en detalle.\",\n        media={\"type\": \"image\", \"format\": fmt, \"data\": image_b64},\n    )\n    print(response)\nelse:\n    print(f\"Image not found: {IMAGE_PATH}\")\n    print(\"Place a test image in this directory and update IMAGE_PATH to test.\")",
 91     "metadata": {},
 92     "execution_count": null,
 93     "outputs": []
 94    },
 95    {
 96     "cell_type": "markdown",
 97     "id": "2qay65ftlal",
 98     "source": "## 6. Test - Audio Transcript",
 99     "metadata": {}
100    },
101    {
102     "cell_type": "code",
103     "id": "dt5qn8xpp3",
104     "source": "response = invoke_agent(\n    prompt=\"Procesa este audio\",\n    media={\n        \"type\": \"audio_transcript\",\n        \"data\": \"Necesito agendar una reunion para el martes a las 3 de la tarde con el equipo de marketing para hablar de la estrategia del Q4.\",\n    },\n)\nprint(response)",
105     "metadata": {},
106     "execution_count": null,
107     "outputs": []
108    },
109    {
110     "cell_type": "markdown",
111     "id": "scw6n059u3s",
112     "source": "## 7. Test - Video Analysis (TwelveLabs Pegasus)\n\n> Upload a test video to S3 first and update `VIDEO_S3_URI`.\n> The TwelveLabs Pegasus model (`twelvelabs.pegasus-1-2-v1:0`) must be enabled in Bedrock.",
113     "metadata": {}
114    },
115    {
116     "cell_type": "code",
117     "id": "i162djhxyp",
118     "source": "# Upload the local video to the stack's S3 bucket\ns3_client = boto3.client(\"s3\", region_name=REGION)\ns3_client.upload_file(\"Runtime.mp4\", S3_BUCKET, \"videos/Runtime.mp4\")\nprint(f\"Uploaded Runtime.mp4 to s3://{S3_BUCKET}/videos/Runtime.mp4\")\n\nVIDEO_S3_URI = f\"s3://{S3_BUCKET}/videos/Runtime.mp4\"\n\nresponse = invoke_agent(\n    prompt=\"Analiza este video en detalle.\",\n    media={\"type\": \"video\", \"s3_uri\": VIDEO_S3_URI},\n)\nprint(response)",
119     "metadata": {},
120     "execution_count": null,
121     "outputs": []
122    },
123    {
124     "cell_type": "markdown",
125     "id": "f284dasfvn",
126     "source": "## 8. Test - Document Processing\n\n> Place a test PDF in this directory and update `DOCUMENT_PATH`.",
127     "metadata": {}
128    },
129    {
130     "cell_type": "code",
131     "id": "tfz6pkrexrk",
132     "source": "DOCUMENT_PATH = \"sample_document.pdf\"  # Change this\n\nif Path(DOCUMENT_PATH).exists():\n    doc_b64 = base64.b64encode(Path(DOCUMENT_PATH).read_bytes()).decode(\"utf-8\")\n    ext = Path(DOCUMENT_PATH).suffix.lstrip(\".\")\n    name = Path(DOCUMENT_PATH).stem\n\n    response = invoke_agent(\n        prompt=\"Resume este documento.\",\n        media={\"type\": \"document\", \"format\": ext, \"data\": doc_b64, \"name\": name},\n    )\n    print(response)\nelse:\n    print(f\"Document not found: {DOCUMENT_PATH}\")\n    print(\"Place a test PDF in this directory and update DOCUMENT_PATH to test.\")",
133     "metadata": {},
134     "execution_count": null,
135     "outputs": []
136    },
137    {
138     "cell_type": "markdown",
139     "id": "qtwbm4kpe7",
140     "source": "## 9. Test - Cross-turn Memory\n\nThe agent should recall information from all previous interactions in this session.",
141     "metadata": {}
142    },
143    {
144     "cell_type": "code",
145     "id": "hkrcv3zu2ug",
146     "source": "# Should recall: name=Carlos, country=Colombia, meeting=Tuesday 3PM marketing Q4\nresponse = invoke_agent(\"Dame un resumen de todo lo que hemos hablado en esta conversacion.\")\nprint(response)",
147     "metadata": {},
148     "execution_count": null,
149     "outputs": []
150    },
151    {
152     "cell_type": "markdown",
153     "id": "ratj2rpiokc",
154     "source": "## 10. Test - Check Memory Records (data plane)\n\nQuery the AgentCore Memory directly to inspect what was stored.",
155     "metadata": {}
156    },
157    {
158     "cell_type": "code",
159     "id": "mslgq8jlu1",
160     "source": "# List memory records for our test actor\n# namespace = actor_id prefix used by the agent\ntry:\n    records = agentcore.list_memory_records(\n        memoryId=MEMORY_ID,\n        namespace=ACTOR_ID,\n    )\n    memory_records = records.get(\"memoryRecords\", records.get(\"records\", []))\n    print(f\"Found {len(memory_records)} memory records:\\n\")\n    for record in memory_records:\n        print(f\"  - Type: {record.get('type', 'N/A')}\")\n        content = record.get(\"content\", record.get(\"text\", \"N/A\"))\n        print(f\"    Content: {str(content)[:200]}\")\n        print()\nexcept Exception as e:\n    print(f\"Error querying memory: {e}\")",
161     "metadata": {},
162     "execution_count": null,
163     "outputs": []
164    }
165   ],
166   "metadata": {
167    "kernelspec": {
168     "display_name": "Python 3",
169     "language": "python",
170     "name": "python3"
171    },
172    "language_info": {
173     "name": "python",
174     "version": "3.11.0"
175    }
176   },
177   "nbformat": 4,
178   "nbformat_minor": 5
179  }