diff --git a/docs/MCP_REGISTRY_DISCOVERY.md b/docs/MCP_REGISTRY_DISCOVERY.md
new file mode 100644
index 00000000..b96fc967
--- /dev/null
+++ b/docs/MCP_REGISTRY_DISCOVERY.md
@@ -0,0 +1,170 @@
+# MCP Server Discovery from AWS Agent Registry
+
+Discover MCP servers from an [AWS Agent Registry](https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/registry.html)
+at agent runtime and **auto-connect** to them, so the registry's approved MCP
+records become live, callable tools on your FAST agent — with **no
+DynamoDB, no UI, and no per-user preferences**. This is a deliberately
+lightweight feature: enable it, point it at a registry, and the agent does the
+rest on each request.
+
+## How it differs from Dynamic MCP Servers
+
+FAST has two, independent ways to add external MCP servers. Pick whichever fits;
+they do not depend on each other.
+
+| | **MCP Registry Discovery** (this doc) | **[Dynamic MCP Servers](MCP_SERVERS.md)** |
+|---|---|---|
+| Source of servers | An AWS Agent Registry, queried at **runtime** | A catalog you declare at **deploy time** in `config.yaml` / `tfvars` |
+| How servers connect | Direct Strands `MCPClient`s on the agent | AgentCore **Gateway targets** |
+| Per-user on/off | No (agent-wide) | Yes (DynamoDB-backed preferences + settings UI) |
+| Extra infrastructure | One IAM statement + two env vars | DynamoDB table, `mcp-prefs` Lambda, API routes, frontend dialog |
+| Best for | "Let the agent use whatever the org has published" | "Curated, per-user-toggleable set of servers" |
+
+## Overview
+
+When enabled, on each request the agent:
+
+1. Lists the registry's **Approved** `recordType == "MCP"` records
+   (`list_discoverable_registry_records`, paginated).
+2. Fetches their full descriptors in one batch call
+   (`batch_get_discoverable_registry_record`) and reads each server's
+   streamable-HTTP endpoint from `mcpServer.remotes[0].url`.
+3. Builds a live Strands `MCPClient` for each **public streamable-HTTP** server
+   and adds it to the agent's tools, alongside the Gateway client and Code
+   Interpreter.
+
+```
+                         ┌──────────────────────────┐
+   Agent Runtime  ──────▶│  AWS Agent Registry       │  list_discoverable_registry_records
+ (basic_agent.py)        │  (agent-registry API)     │  batch_get_discoverable_registry_record
+        │                └──────────────────────────┘
+        │  build_registry_mcp_clients()  → remotes[0].url
+        ▼
+  ┌───────────────┐   streamable HTTP   ┌───────────────────┐
+  │  Strands      │────────────────────▶│ Discovered MCP     │
+  │  MCPClient(s) │                     │ Server (public)    │
+  └───────────────┘                     └───────────────────┘
+```
+
+## Quick Start
+
+### CDK (`infra-cdk/config.yaml`)
+
+```yaml
+backend:
+  mcp_registry:
+    enabled: true
+    registry_id: arn:aws:agent-registry:us-east-1:123456789012:registry/my-registry
+```
+
+Deploy with `cdk deploy`.
+
+### Terraform (`terraform.tfvars`)
+
+```hcl
+mcp_registry = {
+  enabled     = true
+  registry_id = "arn:aws:agent-registry:us-east-1:123456789012:registry/my-registry"
+}
+```
+
+Deploy with `terraform apply`.
+
+## Configuration Reference
+
+| Field | Required | Default | Description |
+|-------|----------|---------|-------------|
+| `enabled` | No | `false` | Master switch. When false the feature is completely inert (no IAM, no env, no runtime calls). |
+| `registry_id` | When enabled | `""` | ARN or id of the AWS Agent Registry. A full ARN scopes the IAM grant to that registry; a bare id falls back to the account/region `registry/*` wildcard. |
+
+Validation is **fail-loud**: enabling the feature without a `registry_id` fails
+at synth/plan time (CDK `config-manager` and the Terraform variable validation).
+
+## Environment Variables (set by the infrastructure)
+
+| Variable | Description |
+|----------|-------------|
+| `MCP_REGISTRY_DISCOVERY_ENABLED` | `"true"` activates discovery in the agent runtime. |
+| `MCP_REGISTRY_ID` | ARN or id of the registry to discover from. |
+| `AWS_REGION` / `AWS_DEFAULT_REGION` | Region of the registry (already set for the runtime). |
+
+## IAM
+
+When enabled, the agent runtime execution role gets two read-only statements.
+**The IAM action names differ from the API names**: the `BatchGetDiscoverableRegistryRecord`
+API is authorized by the permission-only action `agent-registry:GetDiscoverableRegistryRecord`
+on the **record** resource, while `List`/`Search` authorize on the **registry** resource.
+
+```json
+[
+  {
+    "Sid": "AgentRegistryDiscoveryList",
+    "Effect": "Allow",
+    "Action": [
+      "agent-registry:ListDiscoverableRegistryRecords",
+      "agent-registry:SearchDiscoverableRegistryRecords"
+    ],
+    "Resource": "arn:aws:agent-registry:<region>:<account>:registry/<registryId>"
+  },
+  {
+    "Sid": "AgentRegistryDiscoveryGetRecord",
+    "Effect": "Allow",
+    "Action": "agent-registry:GetDiscoverableRegistryRecord",
+    "Resource": "arn:aws:agent-registry:<region>:<account>:registry/<registryId>/record/*"
+  }
+]
+```
+
+> Common pitfall: granting `agent-registry:BatchGetDiscoverableRegistryRecord`
+> (matching the API name) is a **no-op** — that is not a real IAM action, so
+> record reads fail with `AccessDenied`. Use `GetDiscoverableRegistryRecord`.
+
+## Namespace note
+
+The data-plane discovery APIs (`ListDiscoverableRegistryRecords`,
+`BatchGetDiscoverableRegistryRecord`) live under the **`agent-registry`**
+namespace. The public-preview **`bedrock-agentcore`** namespace does **not**
+expose these APIs and is scheduled for discontinuation on **2026-09-17**. This
+feature uses `boto3.client("agent-registry")` accordingly. If you created your
+registry under the old namespace, migrate it first — see the
+[registry migration guide](https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/registry-faq.html).
+
+## Behavior & Limits
+
+- **Only Approved records** are returned by the discovery APIs; Draft /
+  Pending / Rejected / Deprecated records are never connected.
+- **v1 connects public `streamable-http` servers only.** Records using another
+  transport, or requiring authentication, are logged and skipped. Per-server
+  OAuth is a documented follow-up (the [Dynamic MCP Servers](MCP_SERVERS.md)
+  Gateway path already supports OAUTH M2M today if you need auth now).
+- **Fail-loud on misconfiguration**: enabled + no `registry_id` raises at
+  synth/plan time.
+- **Fail-soft at runtime**: if the registry is unreachable, access is denied, or
+  an individual record is malformed, that server is skipped with a logged
+  warning and the agent keeps responding on its remaining tools.
+- **Per-request discovery**: the tool list reflects the registry's current
+  approved records on each new agent invocation.
+
+## Tool Naming
+
+Each discovered server is connected with a Strands client prefix derived from
+its record name: `registry_<slug>`, where `<slug>` is the lowercased name with
+non-alphanumeric runs collapsed to underscores. Its tools therefore appear to
+the agent as `registry_<slug>_<tool_name>`.
+
+## Troubleshooting
+
+| Issue | Cause | Fix |
+|-------|-------|-----|
+| No discovered tools appear | Feature disabled, or registry has no Approved MCP records | Set `enabled: true` + `registry_id`; approve records in the registry |
+| Synth/plan fails on `registry_id` | `enabled` true but `registry_id` empty | Provide the registry ARN/id |
+| A known server isn't connected | Non-HTTP transport, auth-required, or missing `remotes[0].url` | Check the record's descriptor; v1 connects public streamable-HTTP only |
+| `AccessDeniedException` in logs | Role lacks discovery permissions or registry not readable | Confirm the `AgentRegistryDiscoveryAccess` statement covers the registry |
+| APIs return `ValidationException` for the namespace | Registry created under deprecated `bedrock-agentcore` namespace | Migrate the registry to `agent-registry` |
+
+## Files
+
+- `patterns/strands-single-agent/tools/mcp_registry.py` — discovery + client builder
+- `patterns/strands-single-agent/basic_agent.py` — wires discovered clients into the agent
+- `infra-cdk/lib/utils/config-manager.ts`, `infra-cdk/lib/backend-construct.ts`, `infra-cdk/config.yaml` — CDK config, IAM, env
+- `infra-terraform/modules/backend/{variables,runtime}.tf`, `infra-terraform/{variables,main}.tf`, `terraform.tfvars.example` — Terraform parity
diff --git a/infra-cdk/config.yaml b/infra-cdk/config.yaml
index ee1f089e..9ed03a7a 100644
--- a/infra-cdk/config.yaml
+++ b/infra-cdk/config.yaml
@@ -24,6 +24,15 @@ backend:
   ltm_top_k: 10             # Number of facts to retrieve per turn (default: 10)
   ltm_relevance_score: 0.3  # Minimum similarity threshold for retrieval (default: 0.3)
 
+  # MCP server discovery + auto-connect from an AWS Agent Registry.
+  # Lightweight: no DynamoDB, no UI, no per-user preferences. When enabled, the
+  # agent lists the registry's Approved MCP records at runtime and connects to
+  # each public streamable-HTTP server as a live MCP client (their tools become
+  # directly callable). See docs/MCP_REGISTRY_DISCOVERY.md.
+  mcp_registry:
+    enabled: false        # Default off. Set true to activate discovery + auto-connect.
+    registry_id: ""       # ARN or id of the AWS Agent Registry. Required when enabled.
+
   # VPC configuration - required when network_mode is VPC
   # Your VPC must have the necessary VPC endpoints for AWS services.
   # See docs/DEPLOYMENT.md for the full list of required VPC endpoints.
diff --git a/infra-cdk/lib/backend-construct.ts b/infra-cdk/lib/backend-construct.ts
index 5deca4fe..32652e1a 100644
--- a/infra-cdk/lib/backend-construct.ts
+++ b/infra-cdk/lib/backend-construct.ts
@@ -366,6 +366,50 @@ export class BackendConstruct extends Construct {
       })
     )
 
+    // Add AWS Agent Registry discovery access (only when the feature is enabled).
+    // Grants read-only discovery of the registry's Approved MCP records so the
+    // agent can list them and fetch their descriptors to auto-connect at runtime.
+    // Namespace note: the data-plane discovery APIs live under `agent-registry`
+    // (the older `bedrock-agentcore` namespace is deprecated after 2026-09-17).
+    if (config.backend.mcp_registry.enabled) {
+      const registryId = config.backend.mcp_registry.registry_id
+      // Resource ARNs differ by action per the AWS Agent Registry authorization
+      // reference: List/Search authorize on the *registry* resource, while the
+      // BatchGetDiscoverableRegistryRecord API is authorized by the permission-
+      // only action agent-registry:GetDiscoverableRegistryRecord on the *record*
+      // resource (registry/<id>/record/*). Note the API name and the IAM action
+      // name differ — granting "BatchGetDiscoverableRegistryRecord" is a no-op
+      // and yields AccessDenied on record reads.
+      const registryArn = registryId.startsWith("arn:")
+        ? registryId
+        : `arn:aws:agent-registry:${this.region}:${this.account}:registry/*`
+      const recordArn = registryId.startsWith("arn:")
+        ? `${registryId}/record/*`
+        : `arn:aws:agent-registry:${this.region}:${this.account}:registry/*/record/*`
+      // Registry-level: enumerate + natural-language search.
+      agentRole.addToPolicy(
+        new iam.PolicyStatement({
+          sid: "AgentRegistryDiscoveryList",
+          effect: iam.Effect.ALLOW,
+          actions: [
+            "agent-registry:ListDiscoverableRegistryRecords",
+            "agent-registry:SearchDiscoverableRegistryRecords",
+          ],
+          resources: [registryArn],
+        })
+      )
+      // Record-level: read full descriptors via BatchGet (authorized by the
+      // permission-only GetDiscoverableRegistryRecord action on the record ARN).
+      agentRole.addToPolicy(
+        new iam.PolicyStatement({
+          sid: "AgentRegistryDiscoveryGetRecord",
+          effect: iam.Effect.ALLOW,
+          actions: ["agent-registry:GetDiscoverableRegistryRecord"],
+          resources: [recordArn],
+        })
+      )
+    }
+
     // Environment variables for the runtime
     const envVars: { [key: string]: string } = {
       AWS_REGION: stack.region,
@@ -381,6 +425,12 @@ export class BackendConstruct extends Construct {
       // See config.yaml: ltm_top_k and ltm_relevance_score.
       LTM_TOP_K: String(config.backend.ltm_top_k),
       LTM_RELEVANCE_SCORE: String(config.backend.ltm_relevance_score),
+      // Discover + auto-connect MCP servers from an AWS Agent Registry (opt-in).
+      // When enabled, the agent lists the registry's Approved MCP records and
+      // connects to each public streamable-HTTP server at runtime. See
+      // config.yaml: mcp_registry and docs/MCP_REGISTRY_DISCOVERY.md.
+      MCP_REGISTRY_DISCOVERY_ENABLED: config.backend.mcp_registry.enabled ? "true" : "false",
+      MCP_REGISTRY_ID: config.backend.mcp_registry.registry_id,
     }
 
     // Add claude-agent-sdk specific environment variable
diff --git a/infra-cdk/lib/utils/config-manager.ts b/infra-cdk/lib/utils/config-manager.ts
index e17217b4..368335da 100644
--- a/infra-cdk/lib/utils/config-manager.ts
+++ b/infra-cdk/lib/utils/config-manager.ts
@@ -55,9 +55,29 @@ export interface AppConfig {
      * Maps to the relevance_score parameter of RetrievalConfig. Defaults to 0.3.
      */
     ltm_relevance_score: number
+    /**
+     * Discover and auto-connect MCP servers from an AWS Agent Registry.
+     * Lightweight: no DynamoDB, no UI, no per-user preferences. Defaults to disabled.
+     */
+    mcp_registry: McpRegistryConfig
   }
 }
 
+/**
+ * Runtime MCP-server discovery from an AWS Agent Registry.
+ *
+ * When enabled, the agent lists the registry's Approved `recordType=MCP`
+ * records and auto-connects to each public streamable-HTTP server as a live
+ * MCP client. Discovery happens at agent runtime, so no servers are declared
+ * at deploy time and no gateway targets are created.
+ */
+export interface McpRegistryConfig {
+  /** Master switch. When false (default) the feature is completely inert. */
+  enabled: boolean
+  /** ARN or id of the AWS Agent Registry to discover records from. Required when enabled. */
+  registry_id: string
+}
+
 export class ConfigManager {
   private config: AppConfig
 
@@ -137,6 +157,16 @@ export class ConfigManager {
         }
       }
 
+      // Validate MCP registry discovery configuration.
+      // Fail loud: enabling discovery without a registry id is a deploy-time mistake.
+      const mcpRegistryEnabled = parsedConfig.backend?.mcp_registry?.enabled === true
+      const mcpRegistryId = (parsedConfig.backend?.mcp_registry?.registry_id ?? "").trim()
+      if (mcpRegistryEnabled && !mcpRegistryId) {
+        throw new Error(
+          `backend.mcp_registry.registry_id is required in ${configPath} when backend.mcp_registry.enabled is true.`
+        )
+      }
+
       return {
         stack_name_base: stackNameBase,
         admin_user_email: parsedConfig.admin_user_email || null,
@@ -149,6 +179,10 @@ export class ConfigManager {
           use_long_term_memory: parsedConfig.backend?.use_long_term_memory === true,
           ltm_top_k: parsedConfig.backend?.ltm_top_k ?? 10,
           ltm_relevance_score: parsedConfig.backend?.ltm_relevance_score ?? 0.3,
+          mcp_registry: {
+            enabled: mcpRegistryEnabled,
+            registry_id: mcpRegistryId,
+          },
         },
       }
     } catch (error) {
diff --git a/infra-terraform/main.tf b/infra-terraform/main.tf
index 220fe7c2..ae615a88 100644
--- a/infra-terraform/main.tf
+++ b/infra-terraform/main.tf
@@ -95,5 +95,8 @@ module "backend" {
   throttling_rate_limit  = local.api_throttling_rate_limit
   throttling_burst_limit = local.api_throttling_burst_limit
 
+  # MCP server discovery + auto-connect from an AWS Agent Registry (opt-in)
+  mcp_registry = var.mcp_registry
+
   depends_on = [module.cognito, module.amplify_hosting]
 }
diff --git a/infra-terraform/modules/backend/runtime.tf b/infra-terraform/modules/backend/runtime.tf
index 6ee5d66a..9b7aa5aa 100644
--- a/infra-terraform/modules/backend/runtime.tf
+++ b/infra-terraform/modules/backend/runtime.tf
@@ -345,6 +345,45 @@ data "aws_iam_policy_document" "runtime_policy" {
       "arn:aws:bedrock-agentcore:${local.region}:${local.account_id}:workload-identity-directory/*"
     ]
   }
+
+  # AgentRegistryDiscovery: read-only discovery of an AWS Agent Registry's
+  # Approved MCP records so the agent can list them and fetch descriptors to
+  # auto-connect at runtime. Only emitted when the feature is enabled.
+  # Authorization model (per the AWS Agent Registry auth reference):
+  #   - List/Search authorize on the *registry* resource.
+  #   - BatchGetDiscoverableRegistryRecord (the API) is authorized by the
+  #     permission-only action agent-registry:GetDiscoverableRegistryRecord on
+  #     the *record* resource (registry/<id>/record/*). The IAM action name
+  #     differs from the API name — granting "BatchGet..." is a no-op and yields
+  #     AccessDenied on record reads.
+  # Namespace note: these APIs live under `agent-registry` (the older
+  # `bedrock-agentcore` namespace is deprecated after 2026-09-17).
+  dynamic "statement" {
+    for_each = var.mcp_registry.enabled ? [1] : []
+    content {
+      sid    = "AgentRegistryDiscoveryList"
+      effect = "Allow"
+      actions = [
+        "agent-registry:ListDiscoverableRegistryRecords",
+        "agent-registry:SearchDiscoverableRegistryRecords"
+      ]
+      resources = [
+        startswith(var.mcp_registry.registry_id, "arn:") ? var.mcp_registry.registry_id : "arn:aws:agent-registry:${local.region}:${local.account_id}:registry/*"
+      ]
+    }
+  }
+
+  dynamic "statement" {
+    for_each = var.mcp_registry.enabled ? [1] : []
+    content {
+      sid       = "AgentRegistryDiscoveryGetRecord"
+      effect    = "Allow"
+      actions   = ["agent-registry:GetDiscoverableRegistryRecord"]
+      resources = [
+        startswith(var.mcp_registry.registry_id, "arn:") ? "${var.mcp_registry.registry_id}/record/*" : "arn:aws:agent-registry:${local.region}:${local.account_id}:registry/*/record/*"
+      ]
+    }
+  }
 }
 
 resource "aws_iam_role_policy" "runtime" {
@@ -478,6 +517,10 @@ resource "aws_bedrockagentcore_agent_runtime" "main" {
       MEMORY_ID                        = aws_bedrockagentcore_memory.main.id
       STACK_NAME                       = var.stack_name_base
       GATEWAY_CREDENTIAL_PROVIDER_NAME = "${var.stack_name_base}-runtime-gateway-auth"
+      # MCP server discovery + auto-connect from an AWS Agent Registry (opt-in).
+      # See modules/backend/variables.tf: mcp_registry and docs/MCP_REGISTRY_DISCOVERY.md.
+      MCP_REGISTRY_DISCOVERY_ENABLED = var.mcp_registry.enabled ? "true" : "false"
+      MCP_REGISTRY_ID                = var.mcp_registry.registry_id
     },
     # claude-agent-sdk patterns require CLAUDE_CODE_USE_BEDROCK=1
     local.is_claude_agent_sdk ? { CLAUDE_CODE_USE_BEDROCK = "1" } : {}
diff --git a/infra-terraform/modules/backend/variables.tf b/infra-terraform/modules/backend/variables.tf
index 2377e32a..9ce0392c 100644
--- a/infra-terraform/modules/backend/variables.tf
+++ b/infra-terraform/modules/backend/variables.tf
@@ -112,3 +112,29 @@ variable "throttling_burst_limit" {
   default     = 200
 }
 
+# =============================================================================
+# MCP Registry Discovery (lightweight: no DynamoDB, no UI, no per-user prefs)
+# =============================================================================
+
+variable "mcp_registry" {
+  description = <<-EOT
+    Discover and auto-connect MCP servers from an AWS Agent Registry at agent
+    runtime. When enabled, the agent lists the registry's Approved MCP records
+    and connects to each public streamable-HTTP server as a live MCP client.
+    registry_id (ARN or id) is required when enabled.
+  EOT
+  type = object({
+    enabled     = bool
+    registry_id = string
+  })
+  default = {
+    enabled     = false
+    registry_id = ""
+  }
+
+  validation {
+    condition     = !var.mcp_registry.enabled || trimspace(var.mcp_registry.registry_id) != ""
+    error_message = "mcp_registry.registry_id is required when mcp_registry.enabled is true."
+  }
+}
+
diff --git a/infra-terraform/terraform.tfvars.example b/infra-terraform/terraform.tfvars.example
index a6df20ac..2bc2f7bd 100644
--- a/infra-terraform/terraform.tfvars.example
+++ b/infra-terraform/terraform.tfvars.example
@@ -62,3 +62,17 @@ backend_network_mode = "PUBLIC"
 # backend_vpc_id                 = "vpc-xxxxxxxxxxxxxxxxx"
 # backend_vpc_subnet_ids         = ["subnet-xxxxxxxxxxxxxxxxx", "subnet-yyyyyyyyyyyyyyyyy"]
 # backend_vpc_security_group_ids = ["sg-xxxxxxxxxxxxxxxxx"]  # Optional
+
+# -----------------------------------------------------------------------------
+# MCP Registry Discovery + Auto-Connect (optional, lightweight)
+# -----------------------------------------------------------------------------
+# Discover MCP servers from an AWS Agent Registry at agent runtime and connect
+# to each public streamable-HTTP server as a live MCP client (their tools become
+# directly callable). No DynamoDB, no UI, no per-user preferences.
+# See docs/MCP_REGISTRY_DISCOVERY.md. registry_id is required when enabled.
+#
+# mcp_registry = {
+#   enabled     = true
+#   registry_id = "arn:aws:agent-registry:us-east-1:123456789012:registry/my-registry"
+# }
+
diff --git a/infra-terraform/variables.tf b/infra-terraform/variables.tf
index 918aac04..9b89296d 100644
--- a/infra-terraform/variables.tf
+++ b/infra-terraform/variables.tf
@@ -89,3 +89,29 @@ variable "backend_vpc_security_group_ids" {
   default     = []
 }
 
+# =============================================================================
+# MCP Registry Discovery (lightweight: no DynamoDB, no UI, no per-user prefs)
+# =============================================================================
+
+variable "mcp_registry" {
+  description = <<-EOT
+    Discover and auto-connect MCP servers from an AWS Agent Registry at agent
+    runtime. When enabled, the agent lists the registry's Approved MCP records
+    and connects to each public streamable-HTTP server as a live MCP client.
+    registry_id (ARN or id) is required when enabled.
+  EOT
+  type = object({
+    enabled     = bool
+    registry_id = string
+  })
+  default = {
+    enabled     = false
+    registry_id = ""
+  }
+
+  validation {
+    condition     = !var.mcp_registry.enabled || trimspace(var.mcp_registry.registry_id) != ""
+    error_message = "mcp_registry.registry_id is required when mcp_registry.enabled is true."
+  }
+}
+
diff --git a/patterns/strands-single-agent/basic_agent.py b/patterns/strands-single-agent/basic_agent.py
index 426069b5..102e75fa 100644
--- a/patterns/strands-single-agent/basic_agent.py
+++ b/patterns/strands-single-agent/basic_agent.py
@@ -15,6 +15,7 @@
 from strands import Agent
 from strands.models import BedrockModel
 from tools.gateway import create_gateway_mcp_client
+from tools.mcp_registry import build_registry_mcp_clients, is_discovery_enabled
 from utils.auth import extract_user_id_from_context
 
 from tools.code_interpreter import StrandsCodeInterpreterTools
@@ -95,10 +96,28 @@ def create_strands_agent(user_id: str, session_id: str) -> Agent:
 
     gateway_client = create_gateway_mcp_client(user_id)
 
+    # Base tools: Gateway MCP client + secure Code Interpreter.
+    tools: list = [gateway_client, code_tools.execute_python_securely]
+
+    # Auto-connect MCP servers discovered from the AWS Agent Registry (opt-in via
+    # MCP_REGISTRY_DISCOVERY_ENABLED). Each discovered public streamable-HTTP
+    # server becomes a live MCPClient tool provider on the agent. Fail-soft: any
+    # discovery/config error yields zero extra clients and the agent still runs
+    # on its built-in and gateway tools. (Misconfiguration is caught loudly at
+    # deploy time by the CDK config-manager / Terraform variable validation, so
+    # this runtime guard is defense-in-depth, not the primary check.)
+    if is_discovery_enabled():
+        try:
+            tools.extend(build_registry_mcp_clients())
+        except Exception:
+            logger.exception(
+                "[MCP-REGISTRY] Registry discovery failed; continuing without registry tools"
+            )
+
     return Agent(
         name="strands_agent",
         system_prompt=SYSTEM_PROMPT,
-        tools=[gateway_client, code_tools.execute_python_securely],
+        tools=tools,
         model=bedrock_model,
         session_manager=session_manager,
         trace_attributes={"user.id": user_id, "session.id": session_id},
diff --git a/patterns/strands-single-agent/requirements.txt b/patterns/strands-single-agent/requirements.txt
index e29269f8..d9eba7ef 100644
--- a/patterns/strands-single-agent/requirements.txt
+++ b/patterns/strands-single-agent/requirements.txt
@@ -3,3 +3,9 @@ strands-agents==1.32.0
 bedrock-agentcore==1.4.7
 mcp==1.28.1
 PyJWT[crypto]==2.13.0
+# boto3/botocore >=1.43.66 ships the `agent-registry` data-plane client used by
+# tools/mcp_registry.py for MCP server discovery. Pinned so the runtime can call
+# it regardless of what the base image would otherwise resolve. (The discovery
+# module fails soft if the client is unavailable, but pinning makes it work.)
+boto3>=1.43.66
+botocore>=1.43.66
diff --git a/patterns/strands-single-agent/tools/mcp_registry.py b/patterns/strands-single-agent/tools/mcp_registry.py
new file mode 100644
index 00000000..b8dd3821
--- /dev/null
+++ b/patterns/strands-single-agent/tools/mcp_registry.py
@@ -0,0 +1,363 @@
+"""Discover MCP servers from an AWS Agent Registry and auto-connect them.
+
+This module lets the FAST agent discover MCP-server records published in an
+`AWS Agent Registry <https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/registry.html>`_
+at runtime and connect to them as live Strands ``MCPClient`` tool providers, so
+their tools become directly callable by the agent — with no hand-maintained
+catalog, no DynamoDB, and no UI.
+
+Discovery uses the ``agent-registry`` data-plane APIs (the ``bedrock-agentcore``
+namespace is deprecated after 2026-09-17 and does not expose these APIs):
+
+* ``list_discoverable_registry_records`` — paginated summaries of *Approved*
+  records; filtered here to ``recordType == "MCP"``.
+* ``batch_get_discoverable_registry_record`` — full descriptor content, from
+  which the streamable-HTTP endpoint URL is read (``mcpServer.remotes[0].url``).
+
+Environment (set by the infrastructure only when the feature is enabled):
+    MCP_REGISTRY_DISCOVERY_ENABLED  — "true" to activate; anything else disables.
+    MCP_REGISTRY_ID                 — ARN or id of the AWS Agent Registry.
+    AWS_REGION / AWS_DEFAULT_REGION — region of the registry.
+
+Design contract:
+    * Fail loud on misconfiguration: if discovery is enabled but no registry id
+      is configured, :func:`build_registry_mcp_clients` raises ``ValueError`` so
+      the deployment error is obvious rather than silently doing nothing.
+    * Fail soft on runtime/registry errors: if the registry is unreachable or an
+      individual record is malformed, that server is skipped with a logged
+      warning and the agent keeps responding on its remaining tools.
+    * v1 connects only *public* ``streamable-http`` records. Records using any
+      other transport, or advertising authentication requirements, are logged
+      and skipped (per-server OAuth is a documented follow-up).
+"""
+
+import json
+import logging
+import os
+import re
+from dataclasses import dataclass
+
+import boto3
+from mcp.client.streamable_http import streamablehttp_client
+from strands.tools.mcp import MCPClient
+
+logger = logging.getLogger(__name__)
+
+# Registry record filter: only Model Context Protocol server records.
+_MCP_RECORD_TYPE = "MCP"
+
+# The only transport this version can auto-connect. AgentCore Gateway and the
+# registry both speak streamable HTTP MCP; SSE and stdio are not connected here.
+_SUPPORTED_TRANSPORT = "streamable-http"
+
+# Max length of the slug portion of a client prefix. Keeps the full tool name
+# ("registry_<slug>_<tool_name>") within Bedrock's 64-char tool-name limit.
+_MAX_PREFIX_SLUG_LEN = 24
+
+
+@dataclass(frozen=True)
+class DiscoveredMcpServer:
+    """A single MCP server discovered from the AWS Agent Registry.
+
+    Attributes:
+        record_id: The registry record id (or ARN) the server was discovered from.
+        name: Human-readable record name, used to derive a safe client prefix.
+        url: The streamable-HTTP MCP endpoint URL (``remotes[0].url``).
+        transport: The advertised transport type (e.g. ``"streamable-http"``).
+    """
+
+    record_id: str
+    name: str
+    url: str
+    transport: str
+
+
+def is_discovery_enabled() -> bool:
+    """Return whether registry-based MCP discovery is switched on.
+
+    Returns:
+        bool: True only when ``MCP_REGISTRY_DISCOVERY_ENABLED`` is the string
+        ``"true"`` (case-insensitive); False otherwise.
+    """
+    return os.environ.get("MCP_REGISTRY_DISCOVERY_ENABLED", "false").lower() == "true"
+
+
+def _get_registry_id() -> str:
+    """Read and validate the configured registry id.
+
+    Returns:
+        str: The non-empty registry id/ARN from ``MCP_REGISTRY_ID``.
+
+    Raises:
+        ValueError: If discovery is enabled but no registry id is configured.
+            Failing loud here surfaces the deployment mistake instead of
+            silently connecting zero servers.
+    """
+    registry_id = os.environ.get("MCP_REGISTRY_ID", "").strip()
+    if not registry_id:
+        raise ValueError(
+            "MCP_REGISTRY_DISCOVERY_ENABLED is 'true' but MCP_REGISTRY_ID is not set. "
+            "Set the AWS Agent Registry id/ARN or disable discovery."
+        )
+    return registry_id
+
+
+def _registry_client() -> "boto3.client":
+    """Create a boto3 client for the AWS Agent Registry data plane.
+
+    Returns:
+        boto3.client: A client for the ``agent-registry`` service bound to the
+        region from ``AWS_REGION`` / ``AWS_DEFAULT_REGION`` (default us-east-1).
+    """
+    region = os.environ.get("AWS_REGION") or os.environ.get(
+        "AWS_DEFAULT_REGION", "us-east-1"
+    )
+    return boto3.client("agent-registry", region_name=region)
+
+
+def _safe_prefix(name: str) -> str:
+    """Derive a Strands client prefix from a record name.
+
+    Strands prefixes each MCP client's tool names with ``{prefix}_``; the prefix
+    must be a simple token so the resulting tool names stay valid and readable.
+    The slug is length-capped because the full gateway/registry tool name
+    (``{prefix}_{tool_name}``) must stay within Bedrock's 64-character tool-name
+    limit — an over-long registry name would otherwise push tool names past it
+    and the agent would reject them.
+
+    Args:
+        name: The registry record's human-readable name.
+
+    Returns:
+        str: ``registry_<slug>`` where ``<slug>`` is the lowercased name with any
+        run of non-alphanumeric characters collapsed to a single underscore,
+        truncated to keep the prefix short.
+    """
+    slug = re.sub(r"[^a-zA-Z0-9]+", "_", name).strip("_").lower()
+    # Cap the slug so "registry_<slug>_<toolname>" stays well under 64 chars.
+    slug = slug[:_MAX_PREFIX_SLUG_LEN].strip("_")
+    return f"registry_{slug or 'server'}"
+
+
+def _extract_endpoint(record: dict) -> tuple[str | None, str | None]:
+    """Pull the MCP endpoint URL and transport from a full registry record.
+
+    The registry stores the MCP server definition under
+    ``record["descriptors"]["mcpServer"]``. In practice the definition lives in
+    the ``data`` field as a **JSON string** conforming to the MCP server schema,
+    whose ``remotes`` array carries the connection details (only the first remote
+    is used). This helper is tolerant of shape variation: ``data`` may be a JSON
+    string or an already-parsed object, and ``remotes`` may sit directly on the
+    ``mcpServer`` object.
+
+    Args:
+        record: A full record object as returned by
+            ``batch_get_discoverable_registry_record`` (includes ``descriptors``).
+
+    Returns:
+        tuple[str | None, str | None]: ``(url, transport)``. Either element is
+        None when it cannot be located.
+    """
+    if not isinstance(record, dict):
+        return None, None
+
+    # Plural "descriptors" is the real key; tolerate a singular "descriptor" too.
+    descriptors = record.get("descriptors") or record.get("descriptor") or {}
+    mcp_server = (
+        descriptors.get("mcpServer", {}) if isinstance(descriptors, dict) else {}
+    )
+    if not isinstance(mcp_server, dict):
+        return None, None
+
+    # The MCP server definition is normally a JSON string in "data".
+    definition: dict = {}
+    data = mcp_server.get("data")
+    if isinstance(data, str) and data.strip():
+        try:
+            definition = json.loads(data)
+        except (ValueError, TypeError):
+            logger.warning("[MCP-REGISTRY] mcpServer.data is not valid JSON; skipping")
+            return None, None
+    elif isinstance(data, dict):
+        definition = data
+
+    # remotes may live inside the parsed definition or directly on mcp_server.
+    remotes = definition.get("remotes") if isinstance(definition, dict) else None
+    if not remotes:
+        remotes = mcp_server.get("remotes")
+    if not remotes or not isinstance(remotes, list) or not isinstance(remotes[0], dict):
+        return None, None
+
+    first = remotes[0]
+    return first.get("url"), first.get("type")
+
+
+def discover_registry_mcp_servers() -> list[DiscoveredMcpServer]:
+    """Discover approved streamable-HTTP MCP servers from the registry.
+
+    Lists all *Approved* ``recordType == "MCP"`` records (paginated), fetches
+    their full descriptors in a single batch call, and returns one
+    :class:`DiscoveredMcpServer` per record that advertises a usable public
+    streamable-HTTP endpoint. Unusable records (wrong transport, missing URL)
+    are logged and skipped.
+
+    Returns:
+        list[DiscoveredMcpServer]: Discovered, connectable servers. Empty when
+        the feature is disabled, the registry is empty/unreachable, or no record
+        exposes a supported endpoint.
+
+    Raises:
+        ValueError: If discovery is enabled but ``MCP_REGISTRY_ID`` is unset
+            (propagated from :func:`_get_registry_id`).
+    """
+    if not is_discovery_enabled():
+        return []
+
+    registry_id = _get_registry_id()
+
+    try:
+        client = _registry_client()
+        # Collect record ids from the paginated list API. Pages are not dense,
+        # so iterate until no nextToken (the paginator handles that for us).
+        record_ids: list[str] = []
+        paginator = client.get_paginator("list_discoverable_registry_records")
+        for page in paginator.paginate(
+            registryId=registry_id,
+            filters=[{"name": "recordType", "values": [_MCP_RECORD_TYPE]}],
+        ):
+            for record in page.get("registryRecords", []):
+                record_id = record.get("recordId") or record.get("recordArn")
+                if record_id:
+                    record_ids.append(record_id)
+    except Exception:
+        # Fail soft: registry unreachable / access denied / throttled. The agent
+        # continues with its built-in and gateway tools.
+        logger.warning(
+            "[MCP-REGISTRY] Failed to list registry records; skipping discovery",
+            exc_info=True,
+        )
+        return []
+
+    if not record_ids:
+        logger.info(
+            "[MCP-REGISTRY] No approved MCP records found in registry %s", registry_id
+        )
+        return []
+
+    discovered: list[DiscoveredMcpServer] = []
+    try:
+        # batch_get accepts up to 100 record ids per entry; chunk defensively.
+        for chunk_start in range(0, len(record_ids), 100):
+            chunk = record_ids[chunk_start : chunk_start + 100]
+            response = client.batch_get_discoverable_registry_record(
+                entries=[{"registryId": registry_id, "recordIds": chunk}]
+            )
+            for record in response.get("registryRecords", []):
+                server = _to_discovered_server(record)
+                if server is not None:
+                    discovered.append(server)
+            for error in response.get("errors", []):
+                logger.warning(
+                    "[MCP-REGISTRY] Could not fetch record %s: %s",
+                    error.get("recordId"),
+                    error.get("errorCode"),
+                )
+    except Exception:
+        logger.warning(
+            "[MCP-REGISTRY] Failed to batch-get record descriptors; skipping discovery",
+            exc_info=True,
+        )
+        return []
+
+    logger.info(
+        "[MCP-REGISTRY] Discovered %d connectable MCP server(s)", len(discovered)
+    )
+    return discovered
+
+
+def _to_discovered_server(record: dict) -> DiscoveredMcpServer | None:
+    """Convert a full registry record into a connectable server, or skip it.
+
+    Args:
+        record: A full record object (including ``descriptor``) from
+            ``batch_get_discoverable_registry_record``.
+
+    Returns:
+        DiscoveredMcpServer | None: The server when it exposes a public
+        streamable-HTTP endpoint; None (with a logged reason) otherwise.
+    """
+    record_id = record.get("recordId") or record.get("recordArn") or "<unknown>"
+    name = record.get("displayName") or record.get("name") or record_id
+
+    url, transport = _extract_endpoint(record)
+    if not url:
+        logger.warning(
+            "[MCP-REGISTRY] Record %s has no remote endpoint URL; skipping", record_id
+        )
+        return None
+
+    # v1 only auto-connects public streamable-HTTP servers. Transport may be
+    # absent in some descriptors; treat absent as the supported default since
+    # remote HTTP MCP is the registry's primary transport.
+    if transport is not None and transport != _SUPPORTED_TRANSPORT:
+        logger.warning(
+            "[MCP-REGISTRY] Record %s uses unsupported transport '%s'; skipping (only %s is auto-connected)",
+            record_id,
+            transport,
+            _SUPPORTED_TRANSPORT,
+        )
+        return None
+
+    return DiscoveredMcpServer(
+        record_id=record_id,
+        name=name,
+        url=url,
+        transport=transport or _SUPPORTED_TRANSPORT,
+    )
+
+
+def build_registry_mcp_clients() -> list[MCPClient]:
+    """Build live Strands MCP clients for every discovered registry server.
+
+    Each returned client can be dropped directly into a Strands ``Agent``'s
+    ``tools=[...]`` list (a ``MCPClient`` is a tool provider), exactly like the
+    gateway client in :mod:`tools.gateway`. The streamable-HTTP connection is
+    created lazily inside the client's factory lambda so a fresh transport is
+    established on each (re)connection.
+
+    Returns:
+        list[MCPClient]: One client per connectable discovered server. Empty
+        when discovery is disabled or nothing connectable was found.
+
+    Raises:
+        ValueError: If discovery is enabled but ``MCP_REGISTRY_ID`` is unset.
+    """
+    servers = discover_registry_mcp_servers()
+
+    clients: list[MCPClient] = []
+    used_prefixes: set[str] = set()
+    for server in servers:
+        # De-duplicate prefixes: two records whose names slugify identically would
+        # otherwise produce colliding tool-name namespaces. Suffix -2, -3, ... on
+        # collision so each connected server keeps a distinct prefix.
+        base_prefix = _safe_prefix(server.name)
+        prefix = base_prefix
+        n = 2
+        while prefix in used_prefixes:
+            prefix = f"{base_prefix}_{n}"
+            n += 1
+        used_prefixes.add(prefix)
+
+        # Bind the URL per-iteration via a default arg so every lambda captures
+        # its own endpoint (avoids the classic late-binding closure bug).
+        clients.append(
+            MCPClient(
+                lambda url=server.url: streamablehttp_client(url=url),
+                prefix=prefix,
+            )
+        )
+        logger.info(
+            "[MCP-REGISTRY] Connected MCP server '%s' at %s", server.name, server.url
+        )
+
+    return clients
diff --git a/tests/unit/test_mcp_registry.py b/tests/unit/test_mcp_registry.py
new file mode 100644
index 00000000..4f46f7ed
--- /dev/null
+++ b/tests/unit/test_mcp_registry.py
@@ -0,0 +1,340 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""Unit tests for the AWS Agent Registry MCP discovery + auto-connect module.
+
+These tests are self-contained: they inject lightweight stub modules for the
+agent-runtime-only dependencies (``mcp`` and ``strands``) into ``sys.modules``
+before importing the module under test, and mock the boto3 ``agent-registry``
+client. No live AWS calls, no live network, and no requirement that the heavy
+agent dependencies be installed in the test environment.
+"""
+
+import importlib
+import json
+import sys
+import types
+from pathlib import Path
+from unittest import mock
+
+import pytest
+
+# --- Make the strands pattern's package importable as a top-level "tools" pkg ---
+# The module under test lives at patterns/strands-single-agent/tools/mcp_registry.py
+# and imports siblings as ``from tools...`` at runtime (matching the container's
+# working directory). Add that pattern directory to sys.path so ``tools`` resolves.
+_PATTERN_DIR = Path(__file__).resolve().parents[2] / "patterns" / "strands-single-agent"
+
+
+def _install_dependency_stubs() -> None:
+    """Register minimal stubs for ``mcp`` and ``strands`` in sys.modules.
+
+    Only the symbols imported by ``mcp_registry`` are provided:
+    ``mcp.client.streamable_http.streamablehttp_client`` and
+    ``strands.tools.mcp.MCPClient``. Each is a plain callable/class recording its
+    arguments so tests can assert how the module used them.
+    """
+    # mcp.client.streamable_http.streamablehttp_client
+    mcp_pkg = types.ModuleType("mcp")
+    mcp_client_pkg = types.ModuleType("mcp.client")
+    mcp_http_mod = types.ModuleType("mcp.client.streamable_http")
+
+    def _streamablehttp_client(url: str):
+        """Stub transport factory: returns a marker capturing the URL."""
+        return ("streamablehttp_client", url)
+
+    mcp_http_mod.streamablehttp_client = _streamablehttp_client
+    mcp_client_pkg.streamable_http = mcp_http_mod
+    mcp_pkg.client = mcp_client_pkg
+    sys.modules["mcp"] = mcp_pkg
+    sys.modules["mcp.client"] = mcp_client_pkg
+    sys.modules["mcp.client.streamable_http"] = mcp_http_mod
+
+    # strands.tools.mcp.MCPClient
+    strands_pkg = types.ModuleType("strands")
+    strands_tools_pkg = types.ModuleType("strands.tools")
+    strands_mcp_mod = types.ModuleType("strands.tools.mcp")
+
+    class _MCPClient:
+        """Stub MCPClient recording the factory result and prefix."""
+
+        def __init__(self, factory, prefix=None):
+            self.prefix = prefix
+            # Invoke the factory immediately so tests can see the captured URL.
+            self.factory_result = factory()
+
+    strands_mcp_mod.MCPClient = _MCPClient
+    strands_tools_pkg.mcp = strands_mcp_mod
+    strands_pkg.tools = strands_tools_pkg
+    sys.modules["strands"] = strands_pkg
+    sys.modules["strands.tools"] = strands_tools_pkg
+    sys.modules["strands.tools.mcp"] = strands_mcp_mod
+
+
+@pytest.fixture()
+def mcp_registry(monkeypatch):
+    """Import (fresh) the module under test with dependency stubs installed."""
+    _install_dependency_stubs()
+    monkeypatch.syspath_prepend(str(_PATTERN_DIR))
+    # Ensure a clean import each test so module-level state can't leak.
+    sys.modules.pop("tools.mcp_registry", None)
+    module = importlib.import_module("tools.mcp_registry")
+    return importlib.reload(module)
+
+
+def _fake_client(list_pages, batch_response):
+    """Build a mock boto3 agent-registry client.
+
+    Args:
+        list_pages: Iterable of pages returned by the list paginator.
+        batch_response: The dict returned by batch_get_discoverable_registry_record.
+
+    Returns:
+        mock.Mock: A client whose paginator yields ``list_pages`` and whose
+        batch-get returns ``batch_response``.
+    """
+    client = mock.Mock()
+    paginator = mock.Mock()
+    # Return a fresh iterator on every paginate() call so the fake client can be
+    # used by more than one discovery pass (e.g. discover() then build()).
+    paginator.paginate.side_effect = lambda *a, **k: iter(list_pages)
+    client.get_paginator.return_value = paginator
+    client.batch_get_discoverable_registry_record.return_value = batch_response
+    return client
+
+
+# ---------------------------------------------------------------------------
+# is_discovery_enabled
+# ---------------------------------------------------------------------------
+def test_disabled_by_default(mcp_registry, monkeypatch):
+    monkeypatch.delenv("MCP_REGISTRY_DISCOVERY_ENABLED", raising=False)
+    assert mcp_registry.is_discovery_enabled() is False
+
+
+def test_enabled_case_insensitive(mcp_registry, monkeypatch):
+    monkeypatch.setenv("MCP_REGISTRY_DISCOVERY_ENABLED", "TRUE")
+    assert mcp_registry.is_discovery_enabled() is True
+
+
+# ---------------------------------------------------------------------------
+# discover_registry_mcp_servers / build_registry_mcp_clients
+# ---------------------------------------------------------------------------
+def test_disabled_returns_no_servers_and_no_calls(mcp_registry, monkeypatch):
+    monkeypatch.setenv("MCP_REGISTRY_DISCOVERY_ENABLED", "false")
+    with mock.patch.object(mcp_registry, "_registry_client") as client_factory:
+        assert mcp_registry.discover_registry_mcp_servers() == []
+        assert mcp_registry.build_registry_mcp_clients() == []
+        client_factory.assert_not_called()
+
+
+def test_enabled_without_registry_id_fails_loud(mcp_registry, monkeypatch):
+    monkeypatch.setenv("MCP_REGISTRY_DISCOVERY_ENABLED", "true")
+    monkeypatch.delenv("MCP_REGISTRY_ID", raising=False)
+    with pytest.raises(ValueError, match="MCP_REGISTRY_ID"):
+        mcp_registry.discover_registry_mcp_servers()
+
+
+def _mcp_record(record_id, name, url, transport="streamable-http", as_dict=False):
+    """Build a registry record matching the real batch-get shape.
+
+    The registry stores the MCP server definition under
+    ``descriptors.mcpServer.data`` as a JSON **string** (per the live API). Set
+    ``as_dict=True`` to instead place an already-parsed object in ``data`` (the
+    forward-compatible branch the module also handles). ``url``/``transport`` may
+    be None to model a record missing its endpoint.
+    """
+    remotes = []
+    if url is not None:
+        remote = {"url": url}
+        if transport is not None:
+            remote["type"] = transport
+        remotes = [remote]
+    definition = {"name": name, "version": "1.0.0", "remotes": remotes}
+    data = definition if as_dict else json.dumps(definition)
+    return {
+        "recordId": record_id,
+        "displayName": name,
+        "recordType": "MCP",
+        "descriptors": {"mcpServer": {"data": data, "dataSchemaVersion": "2025-12-11"}},
+    }
+
+
+def test_happy_path_discovers_and_builds_clients(mcp_registry, monkeypatch):
+    monkeypatch.setenv("MCP_REGISTRY_DISCOVERY_ENABLED", "true")
+    monkeypatch.setenv("MCP_REGISTRY_ID", "my-registry")
+    monkeypatch.setenv("AWS_REGION", "us-east-1")
+
+    list_pages = [
+        {"registryRecords": [{"recordId": "rec-1", "name": "Weather"}]},
+        {"registryRecords": [{"recordId": "rec-2", "name": "Docs Server"}]},
+    ]
+    batch_response = {
+        "registryRecords": [
+            _mcp_record("rec-1", "Weather", "https://weather.example/mcp"),
+            _mcp_record("rec-2", "Docs Server", "https://docs.example/mcp"),
+        ],
+        "errors": [],
+    }
+    client = _fake_client(list_pages, batch_response)
+
+    with mock.patch.object(mcp_registry, "_registry_client", return_value=client):
+        servers = mcp_registry.discover_registry_mcp_servers()
+        assert [s.url for s in servers] == [
+            "https://weather.example/mcp",
+            "https://docs.example/mcp",
+        ]
+
+        clients = mcp_registry.build_registry_mcp_clients()
+
+    assert len(clients) == 2
+    # Prefixes are slugified and namespaced.
+    assert clients[0].prefix == "registry_weather"
+    assert clients[1].prefix == "registry_docs_server"
+    # Each client's factory captured its own endpoint URL (no closure late-binding bug).
+    assert clients[0].factory_result == (
+        "streamablehttp_client",
+        "https://weather.example/mcp",
+    )
+    assert clients[1].factory_result == (
+        "streamablehttp_client",
+        "https://docs.example/mcp",
+    )
+
+
+def test_duplicate_names_get_distinct_prefixes(mcp_registry, monkeypatch):
+    """Two records that slugify to the same prefix must not collide."""
+    monkeypatch.setenv("MCP_REGISTRY_DISCOVERY_ENABLED", "true")
+    monkeypatch.setenv("MCP_REGISTRY_ID", "my-registry")
+    list_pages = [
+        {
+            "registryRecords": [
+                {"recordId": "r1", "name": "AWS Knowledge"},
+                {"recordId": "r2", "name": "aws-knowledge"},
+            ]
+        }
+    ]
+    batch_response = {
+        "registryRecords": [
+            _mcp_record("r1", "AWS Knowledge", "https://a.example/mcp"),
+            _mcp_record("r2", "aws-knowledge", "https://b.example/mcp"),
+        ],
+        "errors": [],
+    }
+    client = _fake_client(list_pages, batch_response)
+    with mock.patch.object(mcp_registry, "_registry_client", return_value=client):
+        clients = mcp_registry.build_registry_mcp_clients()
+    prefixes = [c.prefix for c in clients]
+    assert len(prefixes) == 2
+    assert len(set(prefixes)) == 2, f"prefixes collided: {prefixes}"
+    assert prefixes[0] == "registry_aws_knowledge"
+    assert prefixes[1] == "registry_aws_knowledge_2"
+
+
+def test_long_name_prefix_is_capped(mcp_registry, monkeypatch):
+    """A very long record name must not blow past Bedrock's tool-name limit."""
+    monkeypatch.setenv("MCP_REGISTRY_DISCOVERY_ENABLED", "true")
+    monkeypatch.setenv("MCP_REGISTRY_ID", "my-registry")
+    long_name = "Super Long AWS Service Documentation And Knowledge MCP Server Name"
+    list_pages = [{"registryRecords": [{"recordId": "rL", "name": long_name}]}]
+    batch_response = {
+        "registryRecords": [_mcp_record("rL", long_name, "https://l.example/mcp")],
+        "errors": [],
+    }
+    client = _fake_client(list_pages, batch_response)
+    with mock.patch.object(mcp_registry, "_registry_client", return_value=client):
+        clients = mcp_registry.build_registry_mcp_clients()
+    # prefix = "registry_" (9) + capped slug (<=24) => <= 33 chars, leaving room
+    # for "_<tool_name>" within Bedrock's 64-char tool-name limit.
+    assert len(clients) == 1
+    assert clients[0].prefix.startswith("registry_")
+    assert (
+        len(clients[0].prefix) <= len("registry_") + mcp_registry._MAX_PREFIX_SLUG_LEN
+    )
+
+
+def test_empty_registry_returns_nothing(mcp_registry, monkeypatch):
+    monkeypatch.setenv("MCP_REGISTRY_DISCOVERY_ENABLED", "true")
+    monkeypatch.setenv("MCP_REGISTRY_ID", "my-registry")
+    client = _fake_client(
+        [{"registryRecords": []}], {"registryRecords": [], "errors": []}
+    )
+    with mock.patch.object(mcp_registry, "_registry_client", return_value=client):
+        assert mcp_registry.discover_registry_mcp_servers() == []
+        # batch-get should not even be attempted when there are no record ids.
+        client.batch_get_discoverable_registry_record.assert_not_called()
+
+
+def test_unsupported_transport_is_skipped(mcp_registry, monkeypatch):
+    monkeypatch.setenv("MCP_REGISTRY_DISCOVERY_ENABLED", "true")
+    monkeypatch.setenv("MCP_REGISTRY_ID", "my-registry")
+    list_pages = [{"registryRecords": [{"recordId": "rec-sse", "name": "SSE"}]}]
+    batch_response = {
+        "registryRecords": [
+            _mcp_record("rec-sse", "SSE", "https://sse.example/mcp", transport="sse"),
+        ],
+        "errors": [],
+    }
+    client = _fake_client(list_pages, batch_response)
+    with mock.patch.object(mcp_registry, "_registry_client", return_value=client):
+        assert mcp_registry.discover_registry_mcp_servers() == []
+
+
+def test_missing_url_is_skipped(mcp_registry, monkeypatch):
+    monkeypatch.setenv("MCP_REGISTRY_DISCOVERY_ENABLED", "true")
+    monkeypatch.setenv("MCP_REGISTRY_ID", "my-registry")
+    list_pages = [{"registryRecords": [{"recordId": "rec-x", "name": "NoUrl"}]}]
+    batch_response = {
+        "registryRecords": [
+            _mcp_record("rec-x", "NoUrl", url=None),
+        ],
+        "errors": [],
+    }
+    client = _fake_client(list_pages, batch_response)
+    with mock.patch.object(mcp_registry, "_registry_client", return_value=client):
+        assert mcp_registry.discover_registry_mcp_servers() == []
+
+
+def test_registry_unreachable_fails_soft(mcp_registry, monkeypatch):
+    monkeypatch.setenv("MCP_REGISTRY_DISCOVERY_ENABLED", "true")
+    monkeypatch.setenv("MCP_REGISTRY_ID", "my-registry")
+    client = mock.Mock()
+    client.get_paginator.side_effect = RuntimeError("network down")
+    with mock.patch.object(mcp_registry, "_registry_client", return_value=client):
+        # Fail-soft: no exception propagates, empty result.
+        assert mcp_registry.discover_registry_mcp_servers() == []
+        assert mcp_registry.build_registry_mcp_clients() == []
+
+
+def test_absent_transport_defaults_to_supported(mcp_registry, monkeypatch):
+    monkeypatch.setenv("MCP_REGISTRY_DISCOVERY_ENABLED", "true")
+    monkeypatch.setenv("MCP_REGISTRY_ID", "my-registry")
+    list_pages = [{"registryRecords": [{"recordId": "rec-a", "name": "Ambient"}]}]
+    batch_response = {
+        "registryRecords": [
+            # No "type" on the remote — treated as the supported default.
+            _mcp_record("rec-a", "Ambient", "https://a.example/mcp", transport=None),
+        ],
+        "errors": [],
+    }
+    client = _fake_client(list_pages, batch_response)
+    with mock.patch.object(mcp_registry, "_registry_client", return_value=client):
+        servers = mcp_registry.discover_registry_mcp_servers()
+    assert len(servers) == 1
+    assert servers[0].transport == "streamable-http"
+
+
+def test_descriptor_data_as_parsed_dict(mcp_registry, monkeypatch):
+    """The module also accepts an already-parsed dict in descriptors.mcpServer.data."""
+    monkeypatch.setenv("MCP_REGISTRY_DISCOVERY_ENABLED", "true")
+    monkeypatch.setenv("MCP_REGISTRY_ID", "my-registry")
+    list_pages = [{"registryRecords": [{"recordId": "rec-d", "name": "DictData"}]}]
+    batch_response = {
+        "registryRecords": [
+            _mcp_record("rec-d", "DictData", "https://dict.example/mcp", as_dict=True),
+        ],
+        "errors": [],
+    }
+    client = _fake_client(list_pages, batch_response)
+    with mock.patch.object(mcp_registry, "_registry_client", return_value=client):
+        servers = mcp_registry.discover_registry_mcp_servers()
+    assert len(servers) == 1
+    assert servers[0].url == "https://dict.example/mcp"