fix(templates): allow actor_id overrides to match harness behavior

src/assets/templates/agent-python-strands/README.md

@@ -16,8 +16,22 @@ file defines a Starlette ASGI app with the Strands Agent SDK running within.
 ## Input Validation
 
 Validate invocation input before forwarding it to Strands. Keep plain prompts typed as strings. If the app accepts a
-caller-supplied message history, retain `strip_trailing_tool_use()`, which normalizes the history tail before
-invoking the agent.
+caller-supplied message history, retain `_strip_trailing_tool_use()` in `parse.py`, which normalizes the history tail
+before invoking the agent.
+
+## Payload
+
+The Runtime accepts a JSON object. `parse_payload()` reads:
+
+- `prompt` (string) — a single user message. Used when `messages` is absent; defaults to `""`.
+- `messages` (array) — a full conversation history (`[{"role": ..., "content": [...]}]`). Takes precedence over `prompt`; trailing `toolUse` blocks are stripped before the agent runs.
+- `actorId` (string, optional) — identifies the end user for Memory scoping (e.g. `/users/{actorId}/facts`). Defaults to `"default"`.
+
+Provide either `prompt` or `messages`. The `session_id` is not in the body — it comes from the `X-Amzn-Bedrock-AgentCore-Runtime-Session-Id` header (`context.session_id`); reuse it to continue a conversation.
+
+```json
+{ "prompt": "What's the weather?", "actorId": "user-123" }
+```
 
 ## Environment Variables
 

src/assets/templates/agent-python-strands/main.py

@@ -1,106 +1,65 @@
-from typing import Any
+from functools import lru_cache
 
 from strands import Agent, tool
 from strands.agent.conversation_manager.null_conversation_manager import NullConversationManager
 from bedrock_agentcore.runtime import BedrockAgentCoreApp
 from model.load import load_model
+from parse import parse_payload
 from memory.session import get_memory_session_manager
 
-app = BedrockAgentCoreApp()
-log = app.logger
-
 DEFAULT_SYSTEM_PROMPT = "You are a helpful assistant. Use tools when appropriate."
 
 # Define a collection of tools used by the model
 tools = []
 
-
 # Define a simple function tool
 @tool
 def add_numbers(a: int, b: int) -> int:
     """Return the sum of two numbers"""
     return a + b
 
-
 tools.append(add_numbers)
 
-
 def _make_conversation_manager():
     return NullConversationManager()
 
-
-def agent_factory():
-    cache = {}
-    def get_or_create_agent(session_id, user_id):
-        key = f"{session_id}/{user_id}"
-        if key not in cache:
-            cache[key] = Agent(
-                model=load_model(),
-                session_manager=get_memory_session_manager(session_id, user_id),
-                conversation_manager=_make_conversation_manager(),
-                system_prompt=DEFAULT_SYSTEM_PROMPT,
-                tools=tools,
-            )
-        return cache[key]
-    return get_or_create_agent
-get_or_create_agent = agent_factory()
-
-
-def strip_trailing_tool_use(messages: Any) -> list[dict]:
-    """Strip toolUse blocks from the tail until the last message has none."""
-    if not isinstance(messages, list):
-        raise ValueError("messages must be a list")
-
-    messages = list(messages)
-    while messages:
-        last = messages[-1]
-        if not isinstance(last, dict):
-            raise ValueError("each message must be an object")
-        original_content = last.get("content", [])
-        if not isinstance(original_content, list) or not all(isinstance(block, dict) for block in original_content):
-            raise ValueError("each message content value must be a list of content blocks")
-
-        content = [block for block in original_content if "toolUse" not in block]
-        if len(content) == len(original_content):
-            break
-        if content:
-            messages[-1] = {**last, "content": content}
-            break
-        messages.pop()
-
-    return messages
-
-
-def _extract_prompt(payload: dict):
-    """Accept a caller-supplied message history or a plain prompt string."""
-    if not isinstance(payload, dict):
-        raise ValueError("payload must be a JSON object")
-    if "messages" in payload:
-        return strip_trailing_tool_use(payload["messages"])
-    prompt = payload.get("prompt", "")
-    if not isinstance(prompt, str):
-        raise ValueError("prompt must be a string")
-    return prompt
-
-
-@app.entrypoint
-async def invoke(payload, context):
-    log.info("Invoking Agent.....")
-
-    session_id = getattr(context, "session_id", "default-session")
-    user_id = getattr(context, "user_id", "default-user")
-    agent = get_or_create_agent(session_id, user_id)
-
-    prompt = _extract_prompt(payload)
-
-    async for event in agent.stream_async(prompt):
-        if not isinstance(event, dict) or "event" not in event:
-            continue
-        cbs = event["event"].get("contentBlockStart")
-        if cbs is not None and not cbs.get("start"):
-            continue
-        yield event
-
-
+@lru_cache(maxsize=128)
+def _get_agent(session_id: str, actor_id: str) -> Agent:
+    """Given a session_id and actor_id, return or construct the corresponding Strands agent.
+    Note: caching helps avoid repeated identity fetches on non-bedrock model loads
+    and supports in-memory session management for local dev."""
+    return Agent(
+        model=load_model(),
+        session_manager=get_memory_session_manager(session_id, actor_id),
+        conversation_manager=_make_conversation_manager(),
+        system_prompt=DEFAULT_SYSTEM_PROMPT,
+        tools=tools,
+    )
+
+def create_app():
+    app = BedrockAgentCoreApp()
+    log = app.logger
+
+    @app.entrypoint
+    async def invoke(payload, context):
+        log.info("Invoking Agent.....")
+
+        session_id = getattr(context, "session_id", None) or "default-session"
+        prompt, actor_id = parse_payload(payload)
+
+        log.info(f"Invoking with session_id={session_id} and actor_id={actor_id}")
+        agent = _get_agent(session_id, actor_id)
+
+        async for event in agent.stream_async(prompt):
+            if not isinstance(event, dict) or "event" not in event:
+                continue
+            cbs = event["event"].get("contentBlockStart")
+            if cbs is not None and not cbs.get("start"):
+                continue
+            yield event
+
+    return app
+
+app = create_app()
 if __name__ == "__main__":
     app.run()

src/assets/templates/agent-python-strands/memory/session.py

@@ -1,5 +1,4 @@
 import os
-import uuid
 from typing import Optional
 
 from bedrock_agentcore.memory.integrations.strands.config import AgentCoreMemoryConfig, RetrievalConfig
@@ -8,15 +7,12 @@
 MEMORY_ID = os.getenv("{{memoryEnvVarName}}")
 REGION = os.getenv("AWS_REGION")
 
-
 def get_memory_session_manager(
-    session_id: Optional[str], actor_id: str
+        session_id: str, actor_id: str,
 ) -> Optional[AgentCoreMemorySessionManager]:
     if not MEMORY_ID:
         return None
 
-    session_id = session_id or uuid.uuid4().hex
-
     retrieval_config = {
         f"/users/{actor_id}/facts": RetrievalConfig(top_k=3, relevance_score=0.5),
         f"/users/{actor_id}/preferences": RetrievalConfig(top_k=3, relevance_score=0.5),

src/assets/templates/agent-python-strands/parse.py

@@ -0,0 +1,41 @@
+from typing import Any
+
+def _strip_trailing_tool_use(messages: Any) -> list[dict]:
+    """Strip toolUse blocks from the tail until the last message has none."""
+    if not isinstance(messages, list):
+        raise ValueError("messages must be a list")
+
+    messages = list(messages)
+    while messages:
+        last = messages[-1]
+        if not isinstance(last, dict):
+            raise ValueError("each message must be an object")
+        original_content = last.get("content", [])
+        if not isinstance(original_content, list) or not all(isinstance(block, dict) for block in original_content):
+            raise ValueError("each message content value must be a list of content blocks")
+
+        content = [block for block in original_content if "toolUse" not in block]
+        if len(content) == len(original_content):
+            break
+        if content:
+            messages[-1] = {**last, "content": content}
+            break
+        messages.pop()
+
+    return messages
+
+def parse_payload(payload: dict):
+    """Accept a caller-supplied message history or a plain prompt string."""
+    if not isinstance(payload, dict):
+        raise ValueError("payload must be a JSON object")
+    actor_id = payload.get("actorId", "default")
+    if not isinstance(actor_id, str):
+        raise ValueError("actorId must be a string")
+    if not actor_id:
+        actor_id = "default"
+    if "messages" in payload:
+        return _strip_trailing_tool_use(payload["messages"]), actor_id
+    prompt = payload.get("prompt", "")
+    if not isinstance(prompt, str):
+        raise ValueError("prompt must be a string")
+    return prompt, actor_id

src/core/project/__snapshots__/manager.test.ts.snap

@@ -48,6 +48,7 @@ exports[`FsProjectManager.create snapshots the Strands project manifest and runt
     "app/agent_python_strands/memory/session.py",
     "app/agent_python_strands/model/__init__.py",
     "app/agent_python_strands/model/load.py",
+    "app/agent_python_strands/parse.py",
     "app/agent_python_strands/pyproject.toml",
   ],
   "memories": [