Add FMKB-via-AgentCore-Gateway sample deployed to AgentCore Runtime

rag/managed-knowledge-bases/03-use-case-example/02-fmkb-with-agentcore-runtime/.env.example

@@ -0,0 +1,23 @@
+# Example environment for the Managed KB + AgentCore Gateway runtime example.
+#
+# The real file (.env.fmkb-gateway) is AUTO-GENERATED by 01-raw-mcp/setup_gateway.py
+# once the gateway + KB target are created — you do not write it by hand.
+# It is gitignored; never commit the populated version (it contains your account ID).
+# This file exists only to document the variables the sample produces and consumes.
+
+# --- Written by 01-raw-mcp/setup_gateway.py, sourced before deploy ---
+export REGION=us-west-2
+export ACCOUNT_ID=111122223333
+export KB_ID=XXXXXXXXXX
+export GATEWAY_ID=fmkb-sample-gateway-xxxxxxxxxx
+export GATEWAY_URL=https://fmkb-sample-gateway-xxxxxxxxxx.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp
+export TARGET_ID=XXXXXXXXXX
+export TARGET_NAME=fmkb-sample-target
+export GATEWAY_ROLE_NAME=fmkb-sample-gateway-role
+
+# --- Optional overrides read by 02-strands-agent/fmkb_gateway_strands.py ---
+# GATEWAY_URL is required at runtime (set above / injected into the runtime env).
+# export AWS_REGION=us-west-2
+# export MODEL_ID=us.anthropic.claude-sonnet-4-5-20250929-v1:0
+# export SYSTEM_PROMPT="You are a knowledge-base assistant. Answer ONLY using the KB tool..."
+# export LOG_LEVEL=INFO

rag/managed-knowledge-bases/03-use-case-example/02-fmkb-with-agentcore-runtime/.gitignore

@@ -0,0 +1,8 @@
+.env
+.env.fmkb-gateway
+.bedrock_agentcore.yaml
+.bedrock_agentcore/
+__pycache__/
+*.pyc
+.venv/
+.pytest_cache/

rag/managed-knowledge-bases/03-use-case-example/02-fmkb-with-agentcore-runtime/01-raw-mcp/README.md

@@ -0,0 +1,23 @@
+# 01-raw-mcp — verify the gateway path without an agent
+
+The smallest possible test of the FMKB-as-MCP-tool plumbing. Run this first when something's broken — it isolates the gateway/IAM layer from the agent layer.
+
+## What it does
+
+1. **`setup_gateway.py`** — creates an IAM role for the gateway, an AgentCore Gateway, and a KB target pointing at the Managed KB you supply with `--kb-id`. Writes the resource ids to `.env.fmkb-gateway` in the repo root.
+2. **`raw_mcp_call.py`** — opens an MCP session against the gateway URL using SigV4-signed streamable HTTP, lists tools (one per gateway target), and calls the `<targetName>___Retrieve` tool with the prompt you pass on the CLI. Prints the raw retrieval results.
+3. **`cleanup.py`** — deletes the target, the gateway, and the gateway role. The KB itself is *not* touched.
+
+## Use it
+
+```bash
+pip install -r ../requirements.txt
+python setup_gateway.py --kb-id <YOUR_KB_ID> --region us-west-2
+source ../.env.fmkb-gateway
+
+python raw_mcp_call.py "What does the knowledge base say about cat food?"
+
+python cleanup.py
+```
+
+Required IAM on your caller principal: ability to create/get/delete IAM roles in the `bedrock-agentcore-*` prefix, plus AgentCore control-plane access. See [Use the AgentCore CLI](https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/runtime-permissions.html#runtime-permissions-cli) for the full policy.

rag/managed-knowledge-bases/03-use-case-example/02-fmkb-with-agentcore-runtime/01-raw-mcp/cleanup.py

@@ -0,0 +1,88 @@
+"""Tear down the gateway, KB target, and gateway IAM role created by setup_gateway.py.
+
+Reads ../.env.fmkb-gateway. Does NOT delete the KB.
+"""
+from __future__ import annotations
+
+import os
+import pathlib
+import sys
+import time
+
+import boto3
+from botocore.exceptions import ClientError
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from utils import gateway as gw  # noqa: E402
+
+
+def load_env() -> None:
+    env = ROOT / ".env.fmkb-gateway"
+    if not env.exists():
+        sys.exit(f"{env} not found; nothing to clean up")
+    for line in env.read_text().splitlines():
+        line = line.strip()
+        if not line or line.startswith("#"):
+            continue
+        if line.startswith("export "):
+            line = line[len("export "):]
+        if "=" not in line:
+            continue
+        k, v = line.split("=", 1)
+        os.environ[k.strip()] = v.strip()
+
+
+def main() -> int:
+    load_env()
+    region = os.environ["REGION"]
+    gateway_id = os.environ["GATEWAY_ID"]
+    target_id = os.environ.get("TARGET_ID")
+    role_name = os.environ.get("GATEWAY_ROLE_NAME")
+
+    iam = boto3.client("iam")
+    control = boto3.client("bedrock-agentcore-control", region_name=region)
+
+    if target_id:
+        try:
+            gw.delete_target(gateway_id, target_id, region)
+            print(f"deleted target {target_id}")
+        except ClientError as e:
+            print(f"delete_gateway_target: {e}")
+
+    # Gateway deletion is rejected while any target is still being torn down,
+    # so wait for the target list to drain before retrying.
+    for _ in range(12):
+        try:
+            remaining = control.list_gateway_targets(
+                gatewayIdentifier=gateway_id
+            ).get("items", [])
+        except ClientError:
+            remaining = []
+        if not remaining:
+            break
+        time.sleep(5)
+    try:
+        gw.delete_gateway(gateway_id, region)
+        print(f"deleted gateway {gateway_id}")
+    except ClientError as e:
+        print(f"delete_gateway: {e}")
+
+    if role_name:
+        try:
+            for p in iam.list_role_policies(RoleName=role_name)["PolicyNames"]:
+                iam.delete_role_policy(RoleName=role_name, PolicyName=p)
+            iam.delete_role(RoleName=role_name)
+            print(f"deleted role {role_name}")
+        except ClientError as e:
+            print(f"delete_role: {e}")
+
+    env = ROOT / ".env.fmkb-gateway"
+    env.unlink(missing_ok=True)
+    print(f"removed {env}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

rag/managed-knowledge-bases/03-use-case-example/02-fmkb-with-agentcore-runtime/01-raw-mcp/raw_mcp_call.py

@@ -0,0 +1,76 @@
+"""Call the FMKB Retrieve tool through the gateway via raw MCP+SigV4.
+
+No agent in the loop. Useful as a smoke test before deploying anything to Runtime.
+"""
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import os
+import sys
+
+from mcp.client.session import ClientSession
+from mcp_proxy_for_aws.client import aws_iam_streamablehttp_client
+
+
+async def run(prompt: str, gateway_url: str, region: str, target_name: str) -> int:
+    tool_name = f"{target_name}___Retrieve"
+    async with aws_iam_streamablehttp_client(
+        endpoint=gateway_url,
+        aws_service="bedrock-agentcore",
+        aws_region=region,
+    ) as (read, write, _):
+        async with ClientSession(read, write) as session:
+            await session.initialize()
+            tools = await session.list_tools()
+            names = [t.name for t in tools.tools]
+            if tool_name not in names:
+                print(f"tool {tool_name!r} not found. Available: {names}", file=sys.stderr)
+                return 2
+            result = await session.call_tool(
+                tool_name, {"retrievalQuery": {"text": prompt}}
+            )
+            for c in result.content:
+                if hasattr(c, "text"):
+                    try:
+                        parsed = json.loads(c.text)
+                    except json.JSONDecodeError:
+                        print(c.text)
+                        continue
+                    for hit in parsed.get("retrievalResults", [])[:5]:
+                        score = hit.get("score")
+                        score_str = f"{score:.4f}" if isinstance(score, (int, float)) else "  n/a"
+                        content = hit.get("content") or {}
+                        # `content` is a tagged union — only TEXT chunks have `.text`.
+                        if content.get("type") == "TEXT" or "text" in content:
+                            text = (content.get("text") or "")[:300].replace("\n", " ")
+                        else:
+                            text = f"<{content.get('type','UNKNOWN')} chunk; not previewable>"
+                        loc = hit.get("location") or {}
+                        print(f"  score={score_str}  {text}…")
+                        if loc:
+                            print(f"    from: {json.dumps(loc)[:200]}")
+            return 0
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("prompt")
+    args = parser.parse_args()
+
+    missing = [k for k in ("GATEWAY_URL", "REGION", "TARGET_NAME")
+               if not os.environ.get(k)]
+    if missing:
+        sys.exit(f"missing env vars {missing}; did you `source .env.fmkb-gateway`?")
+
+    return asyncio.run(run(
+        prompt=args.prompt,
+        gateway_url=os.environ["GATEWAY_URL"],
+        region=os.environ["REGION"],
+        target_name=os.environ["TARGET_NAME"],
+    ))
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

rag/managed-knowledge-bases/03-use-case-example/02-fmkb-with-agentcore-runtime/01-raw-mcp/setup_gateway.py

@@ -0,0 +1,127 @@
+"""Provision an IAM gateway role + an AgentCore Gateway + an FMKB Retrieve target.
+
+Writes resource ids to ../.env.fmkb-gateway so 02-strands-agent can pick them up.
+
+Idempotency: re-running with the same --name-prefix reuses the existing gateway
+role, gateway, and KB target by exact-name match — no orphans on re-run.
+Names are deterministic (no random suffix) so reuse actually triggers.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import pathlib
+import sys
+import time
+
+import boto3
+from botocore.exceptions import ClientError
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from utils.managed_kb import assert_kb_active  # noqa: E402
+from utils import gateway as gw  # noqa: E402
+
+
+def get_or_create_gateway_role(name: str, account_id: str, region: str,
+                               kb_id: str) -> str:
+    iam = boto3.client("iam")
+    trust = json.dumps(gw.gateway_role_trust_policy(account_id, region))
+    perms = json.dumps(
+        gw.gateway_role_permission_policy(account_id, region, kb_id=kb_id)
+    )
+    try:
+        iam.create_role(
+            RoleName=name,
+            AssumeRolePolicyDocument=trust,
+            Description="AgentCore Gateway role for FMKB Retrieve",
+        )
+        print(f"created role {name}")
+    except ClientError as e:
+        if e.response["Error"]["Code"] != "EntityAlreadyExists":
+            raise
+        iam.update_assume_role_policy(RoleName=name, PolicyDocument=trust)
+        print(f"role {name} exists; refreshed trust policy")
+    iam.put_role_policy(
+        RoleName=name, PolicyName=f"{name}-inline", PolicyDocument=perms,
+    )
+    role_arn = iam.get_role(RoleName=name)["Role"]["Arn"]
+    # IAM is eventually consistent; AgentCore will reject the role if it tries to
+    # assume it before it's visible. A short wait avoids "role not found" on first deploy.
+    time.sleep(8)
+    return role_arn
+
+
+def get_or_create_gateway(name: str, role_arn: str, region: str) -> dict:
+    control = boto3.client("bedrock-agentcore-control", region_name=region)
+    for page in control.get_paginator("list_gateways").paginate():
+        for it in page["items"]:
+            if it["name"] == name and it["status"] == "READY":
+                full = control.get_gateway(gatewayIdentifier=it["gatewayId"])
+                print(f"reusing gateway {full['gatewayId']}")
+                return full
+    print(f"creating gateway {name}…")
+    return gw.create_gateway(name=name, role_arn=role_arn, region=region)
+
+
+def get_or_create_target(gateway_id: str, name: str, kb_id: str, region: str) -> dict:
+    control = boto3.client("bedrock-agentcore-control", region_name=region)
+    for it in control.list_gateway_targets(gatewayIdentifier=gateway_id).get("items", []):
+        if it["name"] == name and it["status"] == "READY":
+            print(f"reusing target {it['targetId']}")
+            return control.get_gateway_target(
+                gatewayIdentifier=gateway_id, targetId=it["targetId"]
+            )
+    print(f"creating target {name}…")
+    return gw.create_kb_target(
+        gateway_id=gateway_id, kb_id=kb_id, name=name, region=region,
+    )
+
+
+def main() -> int:
+    p = argparse.ArgumentParser()
+    p.add_argument("--kb-id", required=True, help="Managed KB id (FMKB)")
+    p.add_argument("--region", default="us-west-2")
+    p.add_argument("--name-prefix", default="fmkb-sample",
+                   help="prefix for the gateway role / gateway / target")
+    args = p.parse_args()
+
+    sts = boto3.client("sts").get_caller_identity()
+    account_id = sts["Account"]
+    print(f"caller={sts['Arn']}  account={account_id}  region={args.region}")
+
+    assert_kb_active(args.kb_id, args.region)
+    print(f"KB {args.kb_id} verified ACTIVE / MANAGED")
+
+    # Deterministic names so a second run reuses the same gateway / target /
+    # role instead of creating siblings and orphaning the originals.
+    role_name = f"{args.name_prefix}-gateway-role"
+    gw_name = f"{args.name_prefix}-gateway"
+    target_name = f"{args.name_prefix}-target"
+
+    role_arn = get_or_create_gateway_role(
+        role_name, account_id, args.region, kb_id=args.kb_id,
+    )
+    gw_obj = get_or_create_gateway(gw_name, role_arn, args.region)
+    target = get_or_create_target(gw_obj["gatewayId"], target_name, args.kb_id, args.region)
+
+    env_path = ROOT / ".env.fmkb-gateway"
+    env_path.write_text(
+        "# auto-generated by 01-raw-mcp/setup_gateway.py\n"
+        f"export REGION={args.region}\n"
+        f"export ACCOUNT_ID={account_id}\n"
+        f"export KB_ID={args.kb_id}\n"
+        f"export GATEWAY_ID={gw_obj['gatewayId']}\n"
+        f"export GATEWAY_URL={gw_obj['gatewayUrl']}\n"
+        f"export TARGET_ID={target['targetId']}\n"
+        f"export TARGET_NAME={target['name']}\n"
+        f"export GATEWAY_ROLE_NAME={role_name}\n"
+    )
+    print(f"\nwrote {env_path}")
+    print("source it:  source .env.fmkb-gateway")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())