Merge pull request #12 from co-cddo/ap/feature/agentcore_runtime_base

feat: AgentCore CDK construct with built-in agent template

Author: aakash-cddo

README.md

@@ -8,4 +8,138 @@ This simplifies the deployment of containerised applications in the gds-idea tea
 It is not designed to be used directly but it is a dependency in the [app templates repo](https://github.com/co-cddo/gds-idea-app-templates)
 For instructions on usage please see the docs for gds-idea-app-templates.
 
+## AgentCore
+
+Deploys an [Amazon Bedrock AgentCore](https://docs.aws.amazon.com/bedrock/latest/userguide/agentcore.html) runtime with memory, permissions, and observability pre-configured. The built-in agent uses Strands Agent Framework.
+
+### Quick start (zero-config)
+
+Uses the built-in agent template with sensible defaults — no code to copy:
+
+```python
+from gds_idea_cdk_constructs.agent_core import AgentCore, AgentCoreProperties
+
+AgentCore(
+    app,
+    "MyAgent",
+    props=AgentCoreProperties(runtime_name="my-agent"),
+)
+```
+
+### Built-in agent with custom settings
+
+Configure the model, system prompt, and memory without writing agent code:
+
+```python
+from gds_idea_cdk_constructs.agent_core import (
+    AgentCore,
+    AgentCoreProperties,
+    BuiltInAgent,
+    ModelConfig,
+    MemoryConfig,
+)
+
+AgentCore(
+    app,
+    "MyAgent",
+    props=AgentCoreProperties(
+        runtime_name="my-data-agent",
+        agent=BuiltInAgent(
+            model=ModelConfig(
+                model_id="eu.anthropic.claude-sonnet-4-6",
+                max_tokens=8000,
+                budget_tokens=4000,
+            ),
+            system_prompt="You are a helpful data analyst.",
+            log_level="DEBUG",
+        ),
+        memory=MemoryConfig(name="my-memory"),
+    ),
+)
+```
+
+To disable memory, pass `memory=None`.
+
+### Custom agent code
+
+For full control (adding tools, custom logic), use `CustomAgent`:
+
+```python
+from gds_idea_cdk_constructs.agent_core import (
+    AgentCore,
+    AgentCoreProperties,
+    CustomAgent,
+)
+
+AgentCore(
+    app,
+    "MyAgent",
+    props=AgentCoreProperties(
+        runtime_name="my-agent",
+        agent=CustomAgent(
+            agent_code_directory="my_agent_code/",
+            model_id="eu.anthropic.claude-sonnet-4-6",
+            environment_variables={"MY_API_KEY": "secret"},
+        ),
+        memory=None,
+    ),
+)
+```
+
+Your directory must contain a `Dockerfile` and an `agent.py` entrypoint. The built-in `agent_template/` can be copied as a starting point.
+
+The construct automatically injects these env vars into your container:
+
+| Variable | When |
+|---|---|
+| `MODEL_ID` | Always |
+| `REGION` | Always |
+| `MEMORY_ID` | When `memory` is set |
+
+### Configuration reference
+
+#### `AgentCoreProperties`
+
+| Property | Type | Default | Description |
+|---|---|---|---|
+| `runtime_name` | `str` | *(required)* | Unique name per account/region |
+| `agent` | `BuiltInAgent \| CustomAgent` | `BuiltInAgent()` | Agent mode |
+| `memory` | `MemoryConfig \| None` | `MemoryConfig()` | Memory config, or `None` to skip |
+| `description` | `str` | `"An AgentCore Runtime..."` | Runtime description |
+| `platform` | `Platform` | `LINUX_ARM64` | Docker build target |
+| `removal_policy` | `RemovalPolicy` | `DESTROY` | Removal policy for stateful resources |
+
+#### `BuiltInAgent`
+
+| Property | Type | Default | Description |
+|---|---|---|---|
+| `model` | `ModelConfig` | `ModelConfig()` | Model configuration |
+| `system_prompt` | `str` | `""` | System prompt (overrides default file) |
+| `log_level` | `str` | `"INFO"` | Log level |
+
+#### `ModelConfig`
+
+| Property | Type | Default | Description |
+|---|---|---|---|
+| `model_id` | `str` | `"eu.anthropic.claude-sonnet-4-6"` | Bedrock model ID |
+| `max_tokens` | `int` | `8000` | Max output tokens (thinking + reply) |
+| `budget_tokens` | `int` | `4000` | Thinking budget (must be < max_tokens) |
+| `thinking_enabled` | `bool` | `True` | Enable extended thinking |
+| `max_history` | `int` | `20` | Conversation turns to retain |
+
+#### `CustomAgent`
+
+| Property | Type | Default | Description |
+|---|---|---|---|
+| `agent_code_directory` | `str` | *(required)* | Path to agent code + Dockerfile |
+| `model_id` | `str` | `"eu.anthropic.claude-sonnet-4-6"` | Bedrock model ID |
+| `environment_variables` | `dict` | `{}` | Extra env vars for your container |
+
+#### `MemoryConfig`
+
+| Property | Type | Default | Description |
+|---|---|---|---|
+| `name` | `str` | `"chat_session_store"` | Memory store name |
+| `description` | `str` | `"Stores short-term..."` | Memory store description |
+
 Docs https://co-cddo.github.io/gds-idea-cdk-constructs/

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "gds-idea-cdk-constructs"
-version = "0.4.1"
+version = "0.4.2"
 description = "A repo for commonly used constructs in the team."
 readme = "README.md"
 authors = [
@@ -9,6 +9,7 @@ authors = [
 ]
 requires-python = ">=3.11"
 dependencies = [
+    "aws-cdk-aws-bedrock-agentcore-alpha>=2.251.0a0",
     "aws-cdk-lib>=2.243.0",
     "boto3>=1.35.0",
 ]
@@ -77,6 +78,7 @@ exclude = [
     ".pytest_cache",
     ".venv",
     "venv",
+    "src/gds_idea_cdk_constructs/agent_core/agent_template",
 ]
 
 [tool.ruff.lint]

src/gds_idea_cdk_constructs/agent_core/__init__.py

@@ -0,0 +1,17 @@
+from .props import (
+    AgentCoreProperties,
+    BuiltInAgent,
+    CustomAgent,
+    MemoryConfig,
+    ModelConfig,
+)
+from .stack import AgentCore
+
+__all__ = [
+    "AgentCore",
+    "AgentCoreProperties",
+    "BuiltInAgent",
+    "CustomAgent",
+    "MemoryConfig",
+    "ModelConfig",
+]

src/gds_idea_cdk_constructs/agent_core/agent_template/.dockerignore

@@ -0,0 +1,3 @@
+.venv/
+__pycache__/
+*.pyc

src/gds_idea_cdk_constructs/agent_core/agent_template/Dockerfile

@@ -0,0 +1,25 @@
+FROM ghcr.io/astral-sh/uv:0.9 AS uv
+
+FROM python:3.13-slim
+
+WORKDIR /app
+
+COPY --from=uv /uv /usr/local/bin/uv
+
+RUN apt-get update \
+    && apt-get install -y --no-install-recommends ca-certificates \
+    && rm -rf /var/lib/apt/lists/* \
+    && useradd -m -u 1000 bedrock_agentcore \
+    && chown -R bedrock_agentcore:bedrock_agentcore /app
+
+COPY --chown=bedrock_agentcore:bedrock_agentcore pyproject.toml uv.lock* ./
+
+USER bedrock_agentcore
+
+RUN uv sync --frozen 2>/dev/null || uv sync
+
+COPY --chown=bedrock_agentcore:bedrock_agentcore . .
+
+EXPOSE 8080
+
+CMD ["uv", "run", "python", "agent.py"]

src/gds_idea_cdk_constructs/agent_core/agent_template/_config.py

@@ -0,0 +1,35 @@
+import os
+from dataclasses import dataclass
+from pathlib import Path
+
+
+@dataclass(frozen=True)
+class Config:
+    """Runtime configuration loaded from environment variables."""
+
+    region: str
+    model_id: str
+    memory_id: str | None
+    max_history: int
+    max_tokens: int
+    budget_tokens: int
+    thinking_enabled: bool
+    system_prompt: str
+    actor_id: str = "agent"
+
+    @classmethod
+    def from_env(cls) -> "Config":
+        prompt_file = Path(__file__).parent / "default_system_prompt.md"
+        system_prompt = os.getenv("SYSTEM_PROMPT") or (
+            prompt_file.read_text(encoding="utf-8") if prompt_file.exists() else ""
+        )
+        return cls(
+            region=os.environ["REGION"],
+            model_id=os.environ["MODEL_ID"],
+            memory_id=os.getenv("MEMORY_ID"),
+            max_history=int(os.getenv("MAX_HISTORY", "20")),
+            max_tokens=int(os.getenv("MAX_TOKENS", "8000")),
+            budget_tokens=int(os.getenv("BUDGET_TOKENS", "4000")),
+            thinking_enabled=os.getenv("THINKING_ENABLED", "true").lower() == "true",
+            system_prompt=system_prompt,
+        )

src/gds_idea_cdk_constructs/agent_core/agent_template/_logging.py

@@ -0,0 +1,54 @@
+"""Structured JSON logging for CloudWatch.
+
+Configures stdout-based JSON logging so Docker/CloudWatch captures
+structured output. Must be imported before any other application modules
+to ensure handlers are set before third-party loggers initialise.
+"""
+
+import json
+import logging
+import os
+import sys
+
+LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO").upper()
+
+
+class _JsonFormatter(logging.Formatter):
+    """Formats log records as single-line JSON for CloudWatch Logs Insights."""
+
+    def format(self, record):
+        entry = {
+            "timestamp": self.formatTime(record, self.datefmt),
+            "level": record.levelname,
+            "logger": record.name,
+            "message": record.getMessage(),
+        }
+        if record.exc_info:
+            entry["exception"] = self.formatException(record.exc_info)
+        return json.dumps(entry)
+
+
+def setup_logging() -> logging.Logger:
+    """Configure root logger and return a module-level logger.
+
+    - Clears pre-existing handlers (avoids duplicates from SDK imports).
+    - Writes JSON to stdout (required for Docker log drivers).
+    - Suppresses noisy boto/urllib3 loggers.
+    """
+    level = getattr(logging, LOG_LEVEL, logging.INFO)
+
+    root = logging.getLogger()
+    root.handlers.clear()
+
+    handler = logging.StreamHandler(sys.stdout)
+    handler.setLevel(level)
+    handler.setFormatter(_JsonFormatter(datefmt="%Y-%m-%d %H:%M:%S"))
+
+    root.addHandler(handler)
+    root.setLevel(level)
+
+    # Suppress noisy SDK loggers
+    for name in ("boto3", "botocore", "urllib3"):
+        logging.getLogger(name).setLevel(logging.WARNING)
+
+    return logging.getLogger("agent")

src/gds_idea_cdk_constructs/agent_core/agent_template/_metrics.py

@@ -0,0 +1,70 @@
+"""Token usage extraction and OpenTelemetry reporting."""
+
+import logging
+
+from opentelemetry import trace
+
+logger = logging.getLogger("agent")
+
+
+def _get_attr(obj, key, default=None):
+    """Get a value from a dict or object attribute."""
+    if isinstance(obj, dict):
+        return obj.get(key, default)
+    return getattr(obj, key, default)
+
+
+def extract_and_record_usage(
+    result_obj,
+    session_id: str,
+    model_id: str,
+) -> dict:
+    """Extract token metrics from a Strands result and record to OTel.
+
+    Returns:
+        Dict with keys ``input``, ``output``, ``total`` (all default 0).
+    """
+    usage = {"input": 0, "output": 0, "total": 0}
+
+    try:
+        metrics = _get_attr(result_obj, "metrics")
+        accumulated = _get_attr(metrics, "accumulated_usage") if metrics else None
+
+        if not accumulated:
+            logger.warning("No accumulated_usage in result metrics")
+            return usage
+
+        usage["input"] = accumulated.get("inputTokens", 0)
+        usage["output"] = accumulated.get("outputTokens", 0)
+        usage["total"] = accumulated.get(
+            "totalTokens", usage["input"] + usage["output"]
+        )
+
+        logger.info(
+            "TOKEN_USAGE | Session=%s | In=%d | Out=%d | Total=%d",
+            session_id,
+            usage["input"],
+            usage["output"],
+            usage["total"],
+        )
+
+        _record_otel_span(usage, model_id)
+
+    except Exception:
+        logger.warning("Error extracting Strands metrics", exc_info=True)
+
+    return usage
+
+
+def _record_otel_span(usage: dict, model_id: str) -> None:
+    """Attach token metrics to the current OTel span."""
+    span = trace.get_current_span()
+    if not span.is_recording():
+        logger.warning("OTel span not recording — tokens won't reach dashboard")
+        return
+
+    span.set_attribute("gen_ai.system", "bedrock")
+    span.set_attribute("gen_ai.request.model", model_id)
+    span.set_attribute("gen_ai.usage.input_tokens", usage["input"])
+    span.set_attribute("gen_ai.usage.output_tokens", usage["output"])
+    span.set_attribute("gen_ai.usage.total_tokens", usage["total"])