diff --git a/py/GEMINI.md b/py/GEMINI.md
index 77fec3511e..e07d3a39ad 100644
--- a/py/GEMINI.md
+++ b/py/GEMINI.md
@@ -196,6 +196,7 @@
       with urllib.request.urlopen(url) as response:  # ❌ Blocking!
           return response.read()
 
+
   # CORRECT - non-blocking
   async def fetch_data(url: str) -> bytes:
       async with httpx.AsyncClient() as client:
@@ -211,6 +212,7 @@
       with open(path) as f:  # ❌ Blocking!
           return f.read()
 
+
   # CORRECT - non-blocking
   async def read_file(path: str) -> str:
       async with aiofiles.open(path, encoding='utf-8') as f:
@@ -233,12 +235,14 @@
   ```python
   from genkit.core.http_client import get_cached_client
 
+
   # WRONG - creates new client per request (connection overhead)
   async def call_api(url: str) -> dict:
       async with httpx.AsyncClient() as client:
           response = await client.get(url)
           return response.json()
 
+
   # WRONG - stores client at init time (event loop binding issues)
   class MyPlugin:
       def __init__(self):
@@ -248,6 +252,7 @@
           response = await self._client.get(url)  # May fail in different loop
           return response.json()
 
+
   # CORRECT - uses per-event-loop cached client
   async def call_api(url: str, token: str) -> dict:
       # For APIs with expiring tokens, pass auth headers per-request
@@ -258,6 +263,7 @@
       response = await client.get(url, headers={'Authorization': f'Bearer {token}'})
       return response.json()
 
+
   # CORRECT - for static auth (API keys that don't expire)
   async def call_api_static_auth(url: str) -> dict:
       client = get_cached_client(
@@ -817,9 +823,11 @@ Python-specific development and release scripts:
   ```python
   from pydantic import BaseModel, Field
 
+
   class MyFlowInput(BaseModel):
       prompt: str = Field(default='Hello world', description='User prompt')
 
+
   @ai.flow()
   async def my_flow(input: MyFlowInput) -> str:
       return await ai.generate(prompt=input.prompt)
@@ -831,19 +839,18 @@ Python-specific development and release scripts:
   from typing import Annotated
   from pydantic import Field
 
+
   @ai.flow()
   async def my_flow(
       prompt: Annotated[str, Field(default='Hello world')] = 'Hello world',
-  ) -> str:
-      ...
+  ) -> str: ...
   ```
 
   **Wrong** (defaults won't show in Dev UI):
 
   ```python
   @ai.flow()
-  async def my_flow(prompt: str = 'Hello world') -> str:
-      ...
+  async def my_flow(prompt: str = 'Hello world') -> str: ...
   ```
 
 * **Sample Media URLs**: When samples need to reference an image URL (e.g., for
@@ -879,13 +886,14 @@ Python-specific development and release scripts:
   ```python
   import asyncio
 
+
   async def main():
-    # ...
-    await asyncio.Event().wait()
+      # ...
+      await asyncio.Event().wait()
+
 
   # At the bottom of main.py
   if __name__ == '__main__':
-
       ai.run_main(main())
   ```
 
@@ -977,6 +985,7 @@ When developing Genkit plugins, follow these additional guidelines:
          system: str | None = None  # System prompt override
          metadata: dict[str, Any] | None = None  # Request metadata
 
+
      # Bad: Only basic parameters
      class AnthropicModelConfig(BaseModel):
          temperature: float | None = None
@@ -995,6 +1004,7 @@ When developing Genkit plugins, follow these additional guidelines:
              guardrailVersion: Version of the guardrail (default: "DRAFT").
              performanceConfig: Controls latency optimization settings.
          """
+
          guardrailIdentifier: str | None = None
          guardrailVersion: str | None = None
          performanceConfig: PerformanceConfiguration | None = None
@@ -1066,14 +1076,15 @@ deployment environment. This makes the code more portable and user-friendly glob
   # Good: Named constant with clear purpose
   DEFAULT_OLLAMA_SERVER_URL = 'http://127.0.0.1:11434'
 
+
   class OllamaPlugin:
       def __init__(self, server_url: str | None = None):
           self.server_url = server_url or DEFAULT_OLLAMA_SERVER_URL
 
+
   # Bad: Inline hardcoded value
   class OllamaPlugin:
-      def __init__(self, server_url: str = 'http://127.0.0.1:11434'):
-          ...
+      def __init__(self, server_url: str = 'http://127.0.0.1:11434'): ...
   ```
 
 * **Region-Agnostic Helpers**: For cloud services with regional endpoints, provide helper
@@ -1088,9 +1099,9 @@ deployment environment. This makes the code more portable and user-friendly glob
           raise ValueError('Region is required.')
       # Map region to prefix...
 
+
   # Bad: Hardcoded US default
-  def get_inference_profile_prefix(region: str = 'us-east-1') -> str:
-      ...
+  def get_inference_profile_prefix(region: str = 'us-east-1') -> str: ...
   ```
 
 * **Documentation Examples**: In documentation and docstrings, use placeholder values
@@ -1098,10 +1109,10 @@ deployment environment. This makes the code more portable and user-friendly glob
 
   ```python
   # Good: Clear placeholder
-  endpoint='https://your-resource.openai.azure.com/'
+  endpoint = 'https://your-resource.openai.azure.com/'
 
   # Bad: Looks like it might work
-  endpoint='https://eastus.api.example.com/'
+  endpoint = 'https://eastus.api.example.com/'
   ```
 
 * **What IS Acceptable to Hardcode**:
@@ -1302,6 +1313,7 @@ plugins/{name}/tests/
 ```python
 from unittest.mock import AsyncMock, patch
 
+
 @patch('genkit.plugins.mistral.models.Mistral')
 async def test_generate(mock_client_class):
     mock_client = AsyncMock()
@@ -2306,6 +2318,7 @@ When mocking HTTP clients in tests, mock `get_cached_client` instead of
 ```python
 from unittest.mock import AsyncMock, patch
 
+
 @patch('my_module.get_cached_client')
 async def test_api_call(mock_get_client):
     mock_client = AsyncMock()
@@ -3079,3 +3092,171 @@ done
 
 **Exception:** `bin/install_cli` intentionally omits `pipefail` as it's a user-facing
 install script that handles errors differently for better user experience.
+
+### Session Learnings (2026-02-05): DAP, ASGI Types, and Sample Structure
+
+This session covered several important patterns for Genkit Python development.
+
+#### Dynamic Action Provider (DAP) Best Practices
+
+**1. DAP Tools Are NOT in the Global Registry**
+
+Dynamic tools created via `ai.dynamic_tool()` are intentionally NOT registered in the
+global registry. This means you cannot pass them to `ai.generate(tools=[...])` by name.
+
+```python
+# ❌ WRONG - dynamic tools aren't in the registry
+result = await ai.generate(
+    prompt=query,
+    tools=[t.name for t in dynamic_tools],  # Names won't resolve!
+)
+
+# ✅ CORRECT - invoke dynamic tools directly
+tool = await my_dap.get_action('tool', 'get_weather')
+result = await tool.arun(input)
+```
+
+**2. Combining Multiple DAP Tool Results**
+
+When a query might match multiple tools, collect results instead of returning early:
+
+```python
+# ❌ WRONG - returns after first match
+if tool_a and matches_a:
+    return await tool_a.arun(input)
+if tool_b and matches_b:
+    return await tool_b.arun(input)
+
+# ✅ CORRECT - collect all matching results
+results: list[str] = []
+if tool_a and matches_a:
+    results.append(str((await tool_a.arun(input)).response))
+if tool_b and matches_b:
+    results.append(str((await tool_b.arun(input)).response))
+return ' | '.join(results) if results else 'No matches'
+```
+
+**3. Use asyncio.gather for Concurrent DAP Fetches**
+
+When fetching from multiple DAPs, use `asyncio.gather` for efficiency:
+
+```python
+# ✅ Concurrent - efficient
+weather_cache, finance_cache = await asyncio.gather(
+    weather_dap._cache.get_or_fetch(),  # noqa: SLF001
+    finance_dap._cache.get_or_fetch(),  # noqa: SLF001
+)
+
+# ❌ Sequential - slower
+weather_cache = await weather_dap._cache.get_or_fetch()
+finance_cache = await finance_dap._cache.get_or_fetch()
+```
+
+#### Sample Package Structure
+
+**pyproject.toml `packages` vs Runtime Execution**
+
+The `[tool.hatch.build.targets.wheel].packages` setting is for **wheel building**, not
+runtime execution. Samples should be run directly:
+
+```toml
+# pyproject.toml
+[tool.hatch.build.targets.wheel]
+packages = ["src/dap_demo"]  # For wheel builds
+```
+
+```bash
+# run.sh - direct file execution (NOT -m module)
+uv run src/dap_demo/__init__.py "$@"
+```
+
+When using `-m` module execution, Python requires the module to be in `PYTHONPATH`.
+For samples, direct file execution is simpler and matches other samples.
+
+#### ASGI Type Compatibility
+
+**Protocol-Based Types for Framework Portability**
+
+Use `typing.Protocol` instead of Union types for ASGI compatibility across frameworks:
+
+```python
+# ✅ CORRECT - Protocol-based types work with any ASGI framework
+from typing import Protocol, runtime_checkable
+from collections.abc import Awaitable, Callable, MutableMapping
+
+Scope = MutableMapping[str, Any]
+Receive = Callable[[], Awaitable[MutableMapping[str, Any]]]
+Send = Callable[[MutableMapping[str, Any]], Awaitable[None]]
+
+
+@runtime_checkable
+class ASGIApp(Protocol):
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: ...
+```
+
+**Framework-Specific Middleware Uses Native Types**
+
+When extending framework middleware classes (e.g., Litestar's `AbstractMiddleware`),
+use that framework's native types, not the portable ASGI protocols:
+
+```python
+# For Litestar middleware, use litestar.types
+from litestar.middleware.base import AbstractMiddleware
+from litestar.types import Receive, Scope, Send  # Framework-specific
+
+
+class MyMiddleware(AbstractMiddleware):
+    async def __call__(
+        self,
+        scope: Scope,  # litestar.types.Scope
+        receive: Receive,  # litestar.types.Receive
+        send: Send,  # litestar.types.Send
+    ) -> None: ...
+```
+
+**Application Type Uses Any**
+
+External frameworks define incompatible `Application` types, so use `Any`:
+
+```python
+# Intentional - frameworks have incompatible Application types
+Application = Any
+"""Type alias for ASGI application objects.
+
+Note: Uses Any because external frameworks define their own ASGI types
+that aren't structurally compatible with our Protocol.
+"""
+```
+
+#### Optional Dependencies in Lint Configuration
+
+For optional dependencies used only in type hints, add them to the `lint` dependency
+group rather than using inline ignore comments:
+
+```toml
+# In pyproject.toml [project.optional-dependencies]
+lint = [
+  "litestar>=2.0.0",  # For web/typing.py type resolution
+]
+```
+
+This allows type checkers to resolve imports during CI while keeping the package
+optional for runtime.
+
+#### Documentation Style: Avoid Section Marker Comments
+
+Per GEMINI.md guidelines, avoid boilerplate section marker comments:
+
+```python
+# ❌ WRONG - boilerplate markers
+# =============================================================================
+# ASGI Protocol Types
+# =============================================================================
+
+# ✅ CORRECT - descriptive comment only
+# These Protocol-based types follow the ASGI specification and are compatible
+# with any ASGI framework.
+```
+
+Comments should tell **why**, not **what**. Section markers add visual noise
+without adding information.
diff --git a/py/engdoc/parity-analysis/roadmap.md b/py/engdoc/parity-analysis/roadmap.md
index cd788d392e..02e3e3c44a 100644
--- a/py/engdoc/parity-analysis/roadmap.md
+++ b/py/engdoc/parity-analysis/roadmap.md
@@ -4,7 +4,7 @@ This document organizes the identified gaps into executable milestones with depe
 
 ---
 
-## Current Status (Updated 2026-01-30)
+## Current Status (Updated 2026-02-05)
 
 > [!IMPORTANT]
 > **Overall Parity: ~99% Complete** - Nearly all milestones done!
@@ -20,6 +20,7 @@ This document organizes the identified gaps into executable milestones with depe
 | **M4: Telemetry** | ✅ Complete | RealtimeSpanProcessor, flushTracing, AdjustingTraceExporter, GCP parity |
 | **M5: Advanced** | ✅ Complete | embed_many ✅, define_simple_retriever ✅, define_background_model ✅ |
 | **M6: Media Models** | ✅ Complete | Veo, Lyria, TTS, Gemini Image models |
+| **M7: DAP Core** | ✅ Complete | Dynamic Action Provider core implementation |
 
 ### Remaining Work
 
@@ -27,17 +28,131 @@ This document organizes the identified gaps into executable milestones with depe
 |----------|------|--------|--------|
 | **P0** | Testing Infrastructure (`genkit.testing`) | S | ✅ Complete |
 | **P0** | Context Caching (google-genai) | M | ✅ Complete |
+| **P0** | DAP Core Implementation | M | ✅ Complete |
 | **P1** | `define_background_model()` | M | ✅ Complete |
 | **P1** | Veo support in google-genai plugin | M | ✅ Complete |
 | **P1** | TTS (Text-to-Speech) models | S | ✅ Complete |
 | **P1** | Gemini Image models | S | ✅ Complete |
 | **P1** | Lyria audio generation (Vertex AI) | S | ✅ Complete |
+| **P1** | DAP DevUI Integration (`listResolvableActions`) | M | ✅ Complete |
+| **P1** | DAP Registry Key Parsing | S | ✅ Complete |
 | **P1** | Live/Realtime API | L | ❌ Not Started |
 | **P2** | Multi-agent sample | M | ❌ Not Started |
 | **P2** | MCP sample | M | ❌ Not Started |
 
 ---
 
+## M11: Dynamic Action Provider (DAP) Analysis (2026-02-05)
+
+> [!NOTE]
+> DAP enables external systems (e.g., MCP servers) to provide actions at runtime.
+> **Updated 2026-02-05:** Python implementation now at 100% parity with JS PR #4050.
+
+### JS PR #4050 Alignment (Complete)
+
+The Python DAP implementation has been updated to match the latest JavaScript
+changes from PR #4050 (merged 2026-02-05):
+
+| Change | JS (PR #4050) | Python | Status |
+|--------|---------------|--------|--------|
+| **DAP Action Signature** | `z.void()` input, `z.array(ActionMetadataSchema)` output | `None` input, `list[ActionMetadata]` output | ✅ |
+| **Cache Pattern** | `setDap()` / `setValue()` pattern | `set_dap()` / `set_value()` pattern | ✅ |
+| **transform_dap_value** | Returns flat `ActionMetadata[]` | Returns flat `list[ActionMetadataLike]` | ✅ |
+| **Metadata Format** | Includes `name`, `description` explicitly | Same | ✅ |
+| **Action Internal Logic** | Action calls DAP fn directly, caches result | Same | ✅ |
+
+### Feature Comparison: JS vs Python
+
+```
+┌─────────────────────────────────────────────────────────────────────────────┐
+│                    DAP FEATURE COMPARISON: JS vs PYTHON                      │
+├─────────────────────────────────────────────────────────────────────────────┤
+│                                                                             │
+│ CORE FUNCTIONALITY                                                          │
+├────────────────────────────┬────────────────────┬────────────────────┬──────┤
+│ Feature                    │ JS (Canonical)     │ Python             │Status│
+├────────────────────────────┼────────────────────┼────────────────────┼──────┤
+│ define_dynamic_action_prov │ ✅ dynamic-action- │ ✅ blocks/dap.py   │  ✅  │
+│                            │    provider.ts     │                    │      │
+│ is_dynamic_action_provider │ ✅ Lines 127-131   │ ✅ Lines 406-421   │  ✅  │
+│ DynamicActionProvider class│ ✅ Lines 100-125   │ ✅ Lines 289-403   │  ✅  │
+│ SimpleCache with TTL       │ ✅ Lines 31-98     │ ✅ Lines 175-266   │  ✅  │
+│ transform_dap_value        │ ✅ Lines 150-154   │ ✅ Lines 269-286   │  ✅  │
+├────────────────────────────┴────────────────────┴────────────────────┴──────┤
+│                                                                             │
+│ DAP METHODS                                                                 │
+├────────────────────────────┬────────────────────┬────────────────────┬──────┤
+│ getAction(type, name)      │ ✅ registry lookup │ ✅ Lines 317-336   │  ✅  │
+│ listActionMetadata(t,n)    │ ✅ wildcard/prefix │ ✅ Lines 338-375   │  ✅  │
+│ getActionMetadataRecord(p) │ ✅ reflection API  │ ✅ Lines 377-403   │  ✅  │
+│ invalidateCache()          │ ✅ manual cache    │ ✅ Lines 313-315   │  ✅  │
+│ get_or_fetch(skip_trace)   │ ✅ async fetch     │ ✅ Lines 205-238   │  ✅  │
+├────────────────────────────┴────────────────────┴────────────────────┴──────┤
+│                                                                             │
+│ CACHE CONFIGURATION                                                         │
+├────────────────────────────┬────────────────────┬────────────────────┬──────┤
+│ DapConfig interface        │ ✅ name, desc, ttl │ ✅ DapConfig class │  ✅  │
+│ DapCacheConfig (TTL)       │ ✅ ttlMillis       │ ✅ ttl_millis      │  ✅  │
+│ Default TTL (3000ms)       │ ✅ 3000ms default  │ ✅ 3000ms default  │  ✅  │
+│ Negative TTL (no cache)    │ ✅ ttlMillis < 0   │ ✅ ttl_millis < 0  │  ✅  │
+├────────────────────────────┴────────────────────┴────────────────────┴──────┤
+│                                                                             │
+│ REGISTRY INTEGRATION                                                        │
+├────────────────────────────┬────────────────────┬────────────────────┬──────┤
+│ ActionKind/ActionType      │ ✅ 'dynamic-action │ ✅ DYNAMIC_ACTION_ │  ✅  │
+│                            │    -provider'      │    PROVIDER        │      │
+│ DAP fallback in resolve    │ ✅ getDynamicAction│ ✅ registry.py     │  ✅  │
+│                            │                    │    lines 435-456   │      │
+│ listResolvableActions DAP  │ ✅ Includes DAP    │ ✅ list_resolvable_ │  ✅  │
+│                            │    actions in list │    _actions()      │      │
+│ resolveActionNames (DAP)   │ ✅ Wildcard expand │ ✅ resolve_action_ │  ✅  │
+│                            │                    │    _names()        │      │
+│ parseRegistryKey for DAP   │ ✅ Parses DAP keys │ ✅ parse_registry_ │  ✅  │
+│                            │                    │    _key()          │      │
+├────────────────────────────┴────────────────────┴────────────────────┴──────┤
+│                                                                             │
+│ TEST COVERAGE (13 core tests matching JS exactly + 7 additional)            │
+├────────────────────────────┬────────────────────┬────────────────────┬──────┤
+│ gets specific action       │ ✅ Test exists     │ ✅ dap_test.py     │  ✅  │
+│ lists action metadata      │ ✅ Test exists     │ ✅ dap_test.py     │  ✅  │
+│ caches the actions         │ ✅ Test exists     │ ✅ dap_test.py     │  ✅  │
+│ invalidates the cache      │ ✅ Test exists     │ ✅ dap_test.py     │  ✅  │
+│ respects cache ttl         │ ✅ Test exists     │ ✅ dap_test.py     │  ✅  │
+│ lists with prefix          │ ✅ Test exists     │ ✅ dap_test.py     │  ✅  │
+│ lists exact match          │ ✅ Test exists     │ ✅ dap_test.py     │  ✅  │
+│ gets action metadata rec   │ ✅ Test exists     │ ✅ dap_test.py     │  ✅  │
+│ handles concurrent reqs    │ ✅ Test exists     │ ✅ dap_test.py     │  ✅  │
+│ handles fetch errors       │ ✅ Test exists     │ ✅ dap_test.py     │  ✅  │
+│ skips trace when requested │ ✅ Test exists     │ ✅ via skip flag   │  ✅  │
+│ identifies DAPs            │ ✅ Test exists     │ ✅ dap_test.py     │  ✅  │
+│ Additional Python tests    │ -                  │ ✅ 8 extra tests   │  ✅  │
+└────────────────────────────┴────────────────────┴────────────────────┴──────┘
+```
+
+### DAP Completed Gaps ✅
+
+| Gap | JS Location | Impact | Status |
+|-----|-------------|--------|--------|
+| `listResolvableActions` DAP | `registry.ts:383-398` | DevUI shows DAP actions | ✅ `list_resolvable_actions()` |
+| `resolveActionNames` | `registry.ts:196-212` | Wildcard expansion | ✅ `resolve_action_names()` |
+| `parseRegistryKey` DAP | `registry.ts:96-141` | DAP key format parsing | ✅ `parse_registry_key()` |
+
+### Implementation Notes
+
+**Python DAP Core (Complete - 100% Parity with JS PR #4050):**
+- `py/packages/genkit/src/genkit/blocks/dap.py` - Full implementation
+- `py/packages/genkit/tests/genkit/blocks/dap_test.py` - 23 tests (all passing)
+- Documentation with ELI5 explanations and ASCII diagrams in docstrings
+- Sample: `py/samples/dap-demo/` - Comprehensive demonstration
+
+**Registry Integration (Complete 2026-02-05):**
+- `parse_registry_key()` - Parses DAP-style keys like `/dynamic-action-provider/mcp-host:tool/mytool`
+- `resolve_action_names()` - Expands wildcard keys via DAP
+- `list_resolvable_actions()` - Lists all actions including DAP-provided ones
+- `is_action_type()` - Helper to validate action type strings
+
+---
+
 ## Remaining Gaps (Prioritized)
 
 > [!NOTE]
@@ -47,11 +162,14 @@ This document organizes the identified gaps into executable milestones with depe
 |-----|-------------|----------|--------|
 | **Testing Infrastructure** | JS has `echoModel`, `ProgrammableModel`, `TestAction` for unit testing. | **P0** | ✅ Complete |
 | **Context Caching** | `ai.cacheContent()`, `cachedContent` option in generate | **P0** | ✅ Complete |
+| **DAP Core** | Dynamic Action Provider implementation | **P0** | ✅ Complete |
 | **define_background_model** | Core API for background models (Veo, etc.) | **P1** | ✅ Complete |
 | **Veo plugin support** | Add `veo.py` to google-genai plugin (JS has `veo.ts`) | **P1** | ✅ Complete |
 | **TTS models** | Text-to-speech Gemini models (gemini-*-tts) | **P1** | ✅ Complete |
 | **Gemini Image models** | Native image generation (gemini-*-image) | **P1** | ✅ Complete |
 | **Lyria audio generation** | Audio generation via Vertex AI (lyria-002) | **P1** | ✅ Complete |
+| **DAP DevUI Integration** | `listResolvableActions` includes DAP-provided actions | **P1** | ✅ Complete |
+| **DAP Key Parsing** | `parseRegistryKey` for DAP format (`dap:type/name`) | **P2** | ✅ Complete |
 | **Live/Realtime API** | Google GenAI Live API for real-time streaming | **P1** | ❌ Not Started |
 | **CLI/Tooling Parity** | `genkit` CLI commands and Python project behavior | Medium | ⚠️ Mostly Working |
 | **Error Types** | Python error hierarchy parity check | Low | ⚠️ Needs Review |
diff --git a/py/packages/genkit/src/genkit/blocks/dap.py b/py/packages/genkit/src/genkit/blocks/dap.py
index 23d6ac638a..f817429bf5 100644
--- a/py/packages/genkit/src/genkit/blocks/dap.py
+++ b/py/packages/genkit/src/genkit/blocks/dap.py
@@ -105,6 +105,7 @@ async def get_mcp_tools():
 See Also:
     - MCP Plugin: genkit.plugins.mcp for Model Context Protocol integration
     - JS Implementation: js/core/src/dynamic-action-provider.ts
+    - Sample: py/samples/dap-demo for comprehensive examples
 """
 
 import asyncio
@@ -177,22 +178,25 @@ class SimpleCache:
 
     This cache ensures that concurrent requests for the same data share
     a single fetch operation, preventing thundering herd problems.
+
+    Updated to match JS implementation (PR #4050):
+    - Cache is created before DAP action
+    - set_value method for external value assignment
+    - setDap pattern for deferred DAP reference
     """
 
     def __init__(
         self,
-        dap: 'DynamicActionProvider',
         config: DapConfig,
         dap_fn: DapFn,
     ) -> None:
         """Initialize the cache.
 
         Args:
-            dap: The parent DAP action.
             config: DAP configuration including TTL.
             dap_fn: Function to fetch actions from the external source.
         """
-        self._dap = dap
+        self._dap: DynamicActionProvider | None = None
         self._dap_fn = dap_fn
         self._value: DapValue | None = None
         self._expires_at: float | None = None
@@ -202,6 +206,23 @@ def __init__(
         ttl = config.cache_config.ttl_millis if config.cache_config else None
         self._ttl_millis = 3000 if ttl is None or ttl == 0 else ttl
 
+    def set_dap(self, dap: 'DynamicActionProvider') -> None:
+        """Set the DAP reference (deferred initialization).
+
+        Args:
+            dap: The parent DAP action provider.
+        """
+        self._dap = dap
+
+    def set_value(self, value: DapValue) -> None:
+        """Set cache value externally (called by DAP action).
+
+        Args:
+            value: The DAP value to cache.
+        """
+        self._value = value
+        self._expires_at = time.time() * 1000 + self._ttl_millis
+
     async def get_or_fetch(self, skip_trace: bool = False) -> DapValue:
         """Get cached value or fetch fresh data if stale.
 
@@ -240,6 +261,10 @@ async def get_or_fetch(self, skip_trace: bool = False) -> DapValue:
     async def _do_fetch(self, skip_trace: bool) -> DapValue:
         """Perform the actual fetch operation.
 
+        Updated to match JS implementation (PR #4050):
+        - If DAP is set and not skipping trace, run the action (which sets value)
+        - Otherwise, call dap_fn directly and set value
+
         Args:
             skip_trace: If True, skip running the DAP action.
 
@@ -247,13 +272,15 @@ async def _do_fetch(self, skip_trace: bool) -> DapValue:
             Fresh DAP value.
         """
         try:
-            self._value = await self._dap_fn()
-            self._expires_at = time.time() * 1000 + self._ttl_millis
+            if self._dap is not None and not skip_trace:
+                # Run the DAP action - it calls set_value internally
+                await self._dap.action.arun(None)
+            else:
+                # Direct fetch without tracing
+                self.set_value(await self._dap_fn())
 
-            # Run the DAP action for tracing (unless skipped)
-            if not skip_trace:
-                metadata = transform_dap_value(self._value)
-                await self._dap.action.arun(metadata)
+            if self._value is None:
+                raise ValueError('DAP value is None after fetch')
 
             return self._value
         except Exception:
@@ -266,24 +293,45 @@ def invalidate(self) -> None:
         self._expires_at = None
 
 
-def transform_dap_value(value: DapValue) -> DapMetadata:
-    """Transform DAP value to metadata format for logging.
+def _create_action_metadata(action: Action[Any, Any]) -> dict[str, object]:
+    """Create metadata dict from an Action, with Action properties taking precedence.
+
+    Copies action.metadata first, then overwrites name, description, kind, and
+    schemas from the Action object to ensure they are always correct.
+
+    Args:
+        action: The action to create metadata for.
+
+    Returns:
+        A metadata dictionary suitable for ActionMetadata.
+    """
+    meta: dict[str, object] = dict(action.metadata) if action.metadata else {}
+    meta['name'] = action.name
+    meta['description'] = action.description
+    meta['kind'] = action.kind
+    meta['inputSchema'] = action.input_schema
+    meta['outputSchema'] = action.output_schema
+    return meta
+
+
+def transform_dap_value(value: DapValue) -> list[ActionMetadataLike]:
+    """Transform DAP value to flat list of action metadata.
+
+    Updated to match JS implementation (PR #4050):
+    - Returns flat list instead of grouped dict
+    - Matches ActionMetadataSchema structure
 
     Args:
         value: DAP value with actions.
 
     Returns:
-        DAP metadata with action metadata.
+        Flat list of action metadata.
     """
-    metadata: DapMetadata = {}
-    for action_type, actions in value.items():
-        action_metadata_list: list[ActionMetadataLike] = []
-        for action in actions:
-            # Action.metadata is dict[str, object] which satisfies ActionMetadataLike
-            meta: ActionMetadataLike = action.metadata if action.metadata else {}
-            action_metadata_list.append(meta)
-        metadata[action_type] = action_metadata_list
-    return metadata
+    metadata_list: list[ActionMetadataLike] = []
+    for actions in value.values():
+        for action in actions or []:
+            metadata_list.append(_create_action_metadata(action))
+    return metadata_list
 
 
 class DynamicActionProvider:
@@ -297,18 +345,20 @@ def __init__(
         self,
         action: Action[Any, Any],
         config: DapConfig,
-        dap_fn: DapFn,
+        cache: SimpleCache,
     ) -> None:
         """Initialize the DAP.
 
         Args:
             action: The underlying DAP action.
             config: DAP configuration.
-            dap_fn: Function to fetch actions.
+            cache: The cache instance (created before action).
         """
         self.action = action
         self.config = config
-        self._cache = SimpleCache(self, config, dap_fn)
+        self._cache = cache
+        # Set the DAP reference in cache (deferred init pattern from JS)
+        cache.set_dap(self)
 
     def invalidate_cache(self) -> None:
         """Invalidate the cache, forcing a fresh fetch on next access."""
@@ -359,8 +409,7 @@ async def list_action_metadata(
 
         metadata_list: list[ActionMetadataLike] = []
         for action in actions:
-            meta: ActionMetadataLike = action.metadata if action.metadata else {}
-            metadata_list.append(meta)
+            metadata_list.append(_create_action_metadata(action))
 
         # Match all
         if action_name == '*':
@@ -398,7 +447,7 @@ async def get_action_metadata_record(
                 if not action.name:
                     raise ValueError(f'Invalid metadata when listing dynamic actions from {dap_prefix} - name required')
                 key = f'{dap_prefix}:{action_type}/{action.name}'
-                dap_actions[key] = action.metadata if action.metadata else {}
+                dap_actions[key] = _create_action_metadata(action)
 
         return dap_actions
 
@@ -432,6 +481,11 @@ def define_dynamic_action_provider(
     This is useful for integrating with external systems like MCP servers
     or plugin marketplaces.
 
+    Updated to match JS implementation (PR #4050):
+    - DAP action takes no input (None) and returns list[ActionMetadata]
+    - Action calls the DAP function and caches the result
+    - Cache is created before the action
+
     Args:
         registry: The registry to register the DAP with.
         config: DAP configuration or just a name string.
@@ -476,6 +530,9 @@ async def get_tools():
     # Normalize config
     cfg = DapConfig(name=config) if isinstance(config, str) else config
 
+    # Create cache first (matches JS pattern from PR #4050)
+    cache = SimpleCache(cfg, fn)
+
     # Create metadata with DAP type marker
     action_metadata = {
         **cfg.metadata,
@@ -483,9 +540,12 @@ async def get_tools():
     }
 
     # Define the underlying action
-    # The action itself just returns its input (for logging purposes)
-    async def dap_action(input: DapMetadata) -> DapMetadata:
-        return input
+    # Updated to match JS: takes no input, returns list of action metadata
+    # The action itself calls the DAP function and caches the result
+    async def dap_action(_input: None) -> list[ActionMetadataLike]:
+        dap_value = await fn()
+        cache.set_value(dap_value)
+        return transform_dap_value(dap_value)
 
     action = registry.register_action(
         name=cfg.name,
@@ -496,7 +556,10 @@ async def dap_action(input: DapMetadata) -> DapMetadata:
     )
 
     # Wrap in DynamicActionProvider
-    dap = DynamicActionProvider(action, cfg, fn)
+    dap = DynamicActionProvider(action, cfg, cache)
+
+    # Store reference so Registry.list_actions can access it for DevUI
+    action._dap_instance = dap  # type: ignore[attr-defined]
 
     return dap
 
diff --git a/py/packages/genkit/src/genkit/core/action/__init__.py b/py/packages/genkit/src/genkit/core/action/__init__.py
index 4e47132989..357dde99b3 100644
--- a/py/packages/genkit/src/genkit/core/action/__init__.py
+++ b/py/packages/genkit/src/genkit/core/action/__init__.py
@@ -20,7 +20,7 @@
 from ._key import create_action_key, parse_action_key
 from ._tracing import SpanAttributeValue
 from ._util import parse_plugin_name_from_action_name
-from .types import ActionKind, ActionResponse
+from .types import ActionKind, ActionResponse, is_action_type
 
 __all__ = [
     'Action',
@@ -30,6 +30,7 @@
     'ActionRunContext',
     'SpanAttributeValue',
     'create_action_key',
+    'is_action_type',
     'parse_action_key',
     'parse_plugin_name_from_action_name',
 ]
diff --git a/py/packages/genkit/src/genkit/core/action/types.py b/py/packages/genkit/src/genkit/core/action/types.py
index 9dc9d9f472..83b643ed8f 100644
--- a/py/packages/genkit/src/genkit/core/action/types.py
+++ b/py/packages/genkit/src/genkit/core/action/types.py
@@ -59,6 +59,22 @@ class ActionKind(StrEnum):
     UTIL = 'util'
 
 
+def is_action_type(value: str) -> bool:
+    """Check if a string is a valid ActionKind.
+
+    Args:
+        value: The string to check.
+
+    Returns:
+        True if the value is a valid ActionKind.
+    """
+    try:
+        ActionKind(value)
+        return True
+    except ValueError:
+        return False
+
+
 ResponseT = TypeVar('ResponseT')
 
 
diff --git a/py/packages/genkit/src/genkit/core/registry.py b/py/packages/genkit/src/genkit/core/registry.py
index ac7aa75986..7cf28800cc 100644
--- a/py/packages/genkit/src/genkit/core/registry.py
+++ b/py/packages/genkit/src/genkit/core/registry.py
@@ -30,7 +30,7 @@
 import asyncio
 import threading
 from collections.abc import Awaitable, Callable
-from typing import cast
+from typing import Protocol, cast
 
 from dotpromptz.dotprompt import Dotprompt
 from pydantic import BaseModel
@@ -43,7 +43,7 @@
     SpanAttributeValue,
     parse_action_key,
 )
-from genkit.core.action.types import ActionKind, ActionName
+from genkit.core.action.types import ActionKind, ActionName, is_action_type
 from genkit.core.logging import get_logger
 from genkit.core.plugin import Plugin
 from genkit.core.typing import (
@@ -88,6 +88,127 @@
 )
 
 
+def _is_dap_action(action: Action) -> bool:
+    """Check if an action is a Dynamic Action Provider using duck typing.
+
+    Uses metadata check to avoid circular import with genkit.blocks.dap.
+
+    Args:
+        action: The action to check.
+
+    Returns:
+        True if the action is a DAP.
+    """
+    if hasattr(action, 'metadata') and isinstance(action.metadata, dict):
+        return action.metadata.get('type') == 'dynamic-action-provider'
+    return False
+
+
+class DynamicActionProviderProtocol(Protocol):
+    """Protocol for Dynamic Action Provider interface.
+
+    This protocol defines the interface required by Registry to work with DAPs
+    without creating a circular import with genkit.blocks.dap.
+    """
+
+    async def get_action_metadata_record(self, dap_prefix: str) -> dict[str, dict]:
+        """Get action metadata record for DevUI listing.
+
+        Args:
+            dap_prefix: The DAP prefix (e.g., '/dynamic-action-provider/my-dap').
+
+        Returns:
+            A dictionary mapping action keys to metadata dicts.
+        """
+        ...
+
+    async def list_action_metadata(self, action_type: str, action_name: str) -> list[dict]:
+        """List action metadata matching the type and name pattern.
+
+        Args:
+            action_type: The action type (e.g., 'tool').
+            action_name: The action name pattern (supports '*' wildcard).
+
+        Returns:
+            A list of action metadata dicts.
+        """
+        ...
+
+
+class ParsedRegistryKey(BaseModel):
+    """Parsed registry key containing action type, name, and optional DAP host.
+
+    Registry keys can be in several formats:
+    - Standard: /model/googleai/gemini-2.0-flash
+    - DAP: /dynamic-action-provider/mcp-host:tool/my-tool
+    - Util: /util/generate
+
+    Attributes:
+        action_type: The type of action (e.g., 'model', 'tool').
+        action_name: The name of the action.
+        plugin_name: Optional plugin name for namespaced actions.
+        dynamic_action_host: Optional DAP host name for dynamic actions.
+    """
+
+    action_type: str
+    action_name: str
+    plugin_name: str | None = None
+    dynamic_action_host: str | None = None
+
+
+def parse_registry_key(registry_key: str) -> ParsedRegistryKey | None:
+    """Parse a registry key into its component parts.
+
+    Supports multiple key formats:
+    - DAP format: '/dynamic-action-provider/mcp-host:tool/mytool'
+    - Standard format: '/model/googleai/gemini-2.0-flash'
+    - Prompt format: '/prompt/my-plugin/folder/my-prompt'
+    - Util format: '/util/generate'
+
+    Args:
+        registry_key: The registry key string to parse.
+
+    Returns:
+        ParsedRegistryKey containing the parsed components, or None if invalid.
+    """
+    if registry_key.startswith('/dynamic-action-provider'):
+        # DAP format: '/dynamic-action-provider/mcp-host:tool/mytool' or 'mcp-host:tool/*'
+        key_tokens = registry_key.split(':', 1)
+        host_tokens = key_tokens[0].split('/')
+        if len(host_tokens) < 3:
+            return None
+        if len(key_tokens) < 2:
+            return ParsedRegistryKey(
+                action_type=ActionKind.DYNAMIC_ACTION_PROVIDER,
+                action_name=host_tokens[2],
+            )
+        tokens = key_tokens[1].split('/')
+        if len(tokens) < 2 or not is_action_type(tokens[0]):
+            return None
+        return ParsedRegistryKey(
+            dynamic_action_host=host_tokens[2],
+            action_type=tokens[0],
+            action_name='/'.join(tokens[1:]),
+        )
+
+    tokens = registry_key.split('/')
+    if len(tokens) < 3:
+        # Invalid key format
+        return None
+    # Format: /model/googleai/gemini-2.0-flash or /prompt/my-plugin/folder/my-prompt
+    if len(tokens) >= 4:
+        return ParsedRegistryKey(
+            action_type=tokens[1],
+            plugin_name=tokens[2],
+            action_name='/'.join(tokens[3:]),
+        )
+    # Format: /util/generate
+    return ParsedRegistryKey(
+        action_type=tokens[1],
+        action_name=tokens[2],
+    )
+
+
 class Registry:
     """Central repository for Genkit resources.
 
@@ -456,6 +577,78 @@ async def resolve_action(self, kind: ActionKind, name: str) -> Action | None:
 
         return None
 
+    async def resolve_action_names(self, key: str) -> list[str]:
+        """Resolve a registry key to a list of matching action names.
+
+        Supports wildcard expansion for DAP actions
+        (e.g., '/dynamic-action-provider/mcp-host:tool/*').
+
+        Args:
+            key: The registry key, potentially with wildcards.
+
+        Returns:
+            List of fully-qualified action names matching the key.
+        """
+        parsed_key = parse_registry_key(key)
+        if parsed_key and parsed_key.dynamic_action_host:
+            # DAP key - resolve via the DAP
+            host_id = f'/dynamic-action-provider/{parsed_key.dynamic_action_host}'
+            with self._lock:
+                dap_entries = self._entries.get(ActionKind.DYNAMIC_ACTION_PROVIDER, {})
+                dap_action = dap_entries.get(parsed_key.dynamic_action_host)
+
+            if dap_action is None:
+                return []
+
+            if not _is_dap_action(dap_action):
+                return []
+
+            # Get the DynamicActionProvider wrapper (uses Protocol for type hint)
+            dap: DynamicActionProviderProtocol | None = getattr(dap_action, '_dap_instance', None)
+            if dap is None:
+                return []
+
+            metadata_list = await dap.list_action_metadata(parsed_key.action_type, parsed_key.action_name)
+            return [f'{host_id}:{parsed_key.action_type}/{m.get("name", "")}' for m in metadata_list]
+
+        # Standard key - just return it if the action exists
+        if await self.lookup_action_by_key(key):
+            return [key]
+        return []
+
+    async def lookup_action_by_key(self, key: str) -> Action | None:
+        """Lookup an action by its full registry key.
+
+        This is a simple lookup that doesn't trigger any resolution.
+        Use resolve_action_by_key for full resolution with plugin initialization.
+
+        Args:
+            key: The full registry key.
+
+        Returns:
+            The Action if found, None otherwise.
+        """
+        parsed = parse_registry_key(key)
+        if not parsed:
+            return None
+
+        try:
+            kind = ActionKind(parsed.action_type)
+        except ValueError:
+            return None
+
+        if parsed.plugin_name:
+            name = f'{parsed.plugin_name}/{parsed.action_name}'
+        else:
+            name = parsed.action_name
+
+        with self._lock:
+            if kind not in self._entries:
+                return None
+            # pyrefly: ignore[bad-index] - kind is ActionKind, not plain StrEnum
+            kind_entries = self._entries[kind]
+            return kind_entries.get(name)
+
     async def resolve_action_by_key(self, key: str) -> Action | None:
         """Resolve an action using its combined key string.
 
@@ -482,6 +675,9 @@ async def list_actions(self, allowed_kinds: list[ActionKind] | None = None) -> l
         plugins. It does NOT trigger plugin initialization and does NOT consult
         the registry's internal action store.
 
+        For a full list including registered actions and DAP-provided actions,
+        use `list_resolvable_actions` instead.
+
         Args:
             allowed_kinds: Optional list of action kinds to filter by.
 
@@ -509,6 +705,86 @@ async def list_actions(self, allowed_kinds: list[ActionKind] | None = None) -> l
                 metas.append(meta)
         return metas
 
+    async def list_resolvable_actions(self, allowed_kinds: list[ActionKind] | None = None) -> list[ActionMetadata]:
+        """List all resolvable actions including registered and DAP-provided actions.
+
+        This method returns a comprehensive list of all actions:
+        1. Actions advertised by plugins (via list_actions)
+        2. Actions already registered in the registry
+        3. Actions provided by Dynamic Action Providers (DAPs)
+
+        This is the Python equivalent of JS `listResolvableActions` and is used
+        by the DevUI to show all available actions.
+
+        Args:
+            allowed_kinds: Optional list of action kinds to filter by.
+
+        Returns:
+            A list of ActionMetadata objects describing available actions.
+
+        Raises:
+            ValueError: If a plugin returns invalid ActionMetadata.
+        """
+        metas: list[ActionMetadata] = []
+        seen_names: set[str] = set()
+
+        # Get plugin actions first
+        plugin_metas = await self.list_actions(allowed_kinds)
+        for meta in plugin_metas:
+            if meta.name not in seen_names:
+                metas.append(meta)
+                seen_names.add(meta.name)
+
+        # Get all registered actions including DAP actions
+        with self._lock:
+            all_entries = {k: dict(v) for k, v in self._entries.items()}
+
+        for _kind, actions_dict in all_entries.items():
+            for _name, action in actions_dict.items():
+                # Add static action metadata if not already seen
+                if action.name not in seen_names:
+                    # Build metadata dict with all available fields from the action.
+                    # Using model_validate ensures we include schemas for DevUI.
+                    meta_dict: dict[str, object] = {
+                        'name': action.name,
+                        'kind': action.kind,
+                        'description': action.description,
+                        'inputSchema': action.input_schema,
+                        'outputSchema': action.output_schema,
+                    }
+                    # Include any additional metadata from the action
+                    if action.metadata:
+                        meta_dict['metadata'] = action.metadata
+                    meta = ActionMetadata.model_validate(meta_dict)
+                    if allowed_kinds and meta.kind not in allowed_kinds:
+                        continue
+                    metas.append(meta)
+                    seen_names.add(action.name)
+
+                # If this is a DAP, include its dynamic actions
+                if _is_dap_action(action):
+                    try:
+                        dap_prefix = f'/{action.kind}/{action.name}'
+                        # Get the DynamicActionProvider wrapper (uses Protocol for type hint)
+                        dap: DynamicActionProviderProtocol | None = getattr(action, '_dap_instance', None)
+                        if dap:
+                            dap_metadata = await dap.get_action_metadata_record(dap_prefix)
+                            for _dap_key, dap_meta in dap_metadata.items():
+                                if _dap_key not in seen_names:
+                                    # Create a mutable copy to set the fully-qualified name.
+                                    # Using model_validate preserves all metadata fields like schemas.
+                                    meta_dict = dict(dap_meta)
+                                    meta_dict['name'] = _dap_key
+                                    dap_action_meta = ActionMetadata.model_validate(meta_dict)
+                                    if allowed_kinds and dap_action_meta.kind not in allowed_kinds:
+                                        continue
+                                    metas.append(dap_action_meta)
+                                    seen_names.add(_dap_key)
+                    except Exception as e:
+                        logger.error(f'Error listing actions for DAP {action.name}: {e}')
+
+        return metas
+
     def register_schema(self, name: str, schema: dict[str, object], schema_type: type[BaseModel] | None = None) -> None:
         """Registers a schema by name.
 
diff --git a/py/packages/genkit/tests/genkit/blocks/dap_test.py b/py/packages/genkit/tests/genkit/blocks/dap_test.py
index 51e52e0d64..d57eed5a83 100644
--- a/py/packages/genkit/tests/genkit/blocks/dap_test.py
+++ b/py/packages/genkit/tests/genkit/blocks/dap_test.py
@@ -56,7 +56,8 @@
 )
 from genkit.core.action import Action
 from genkit.core.action.types import ActionKind
-from genkit.core.registry import Registry
+from genkit.core.error import GenkitError
+from genkit.core.registry import Registry, parse_registry_key
 
 
 @pytest.fixture
@@ -147,8 +148,9 @@ async def dap_fn() -> DapValue:
 
     metadata = await dap.list_action_metadata('tool', '*')
     assert len(metadata) == 2
-    assert metadata[0] == tool1.metadata
-    assert metadata[1] == tool2.metadata
+    # New API includes name/description in returned metadata
+    assert metadata[0].get('name') == 'tool1'
+    assert metadata[1].get('name') == 'tool2'
     assert call_count == 1
 
 
@@ -248,8 +250,9 @@ async def dap_fn() -> DapValue:
 
     metadata = await dap.list_action_metadata('tool', 'tool*')
     assert len(metadata) == 2
-    assert metadata[0] == tool1.metadata
-    assert metadata[1] == tool2.metadata
+    # New API includes name/description in returned metadata
+    assert metadata[0].get('name') == 'tool1'
+    assert metadata[1].get('name') == 'tool2'
     assert call_count == 1
 
 
@@ -270,7 +273,8 @@ async def dap_fn() -> DapValue:
 
     metadata = await dap.list_action_metadata('tool', 'tool1')
     assert len(metadata) == 1
-    assert metadata[0] == tool1.metadata
+    # New API includes name/description in returned metadata
+    assert metadata[0].get('name') == 'tool1'
     assert call_count == 1
 
 
@@ -296,9 +300,10 @@ async def dap_fn() -> DapValue:
     assert 'dap/my-dap:tool/tool1' in record
     assert 'dap/my-dap:tool/tool2' in record
     assert 'dap/my-dap:flow/tool1' in record
-    assert record['dap/my-dap:tool/tool1'] == tool1.metadata
-    assert record['dap/my-dap:tool/tool2'] == tool2.metadata
-    assert record['dap/my-dap:flow/tool1'] == tool1.metadata
+    # New API returns dict with name/description included
+    assert record['dap/my-dap:tool/tool1'].get('name') == 'tool1'
+    assert record['dap/my-dap:tool/tool2'].get('name') == 'tool2'
+    assert record['dap/my-dap:flow/tool1'].get('name') == 'tool1'
     assert call_count == 1
 
 
@@ -327,18 +332,16 @@ async def dap_fn() -> DapValue:
     metadata1, metadata2 = results
     assert len(metadata1) == 2
     assert len(metadata2) == 2
-    assert metadata1[0] == tool1.metadata
-    assert metadata2[0] == tool1.metadata
+    # New API includes name/description in returned metadata
+    assert metadata1[0].get('name') == 'tool1'
+    assert metadata2[0].get('name') == 'tool1'
     # Only one fetch should have occurred
     assert call_count == 1
 
 
 @pytest.mark.asyncio
 async def test_handles_fetch_errors(registry: Registry, tool1: Action, tool2: Action) -> None:
-    """Test error handling and cache invalidation on fetch failure.
-
-    Corresponds to JS test: 'handles fetch errors'
-    """
+    """Test that DAP raises GenkitError on fetch failure."""
     call_count = 0
 
     async def dap_fn() -> DapValue:
@@ -350,8 +353,8 @@ async def dap_fn() -> DapValue:
 
     dap = define_dynamic_action_provider(registry, 'my-dap', dap_fn)
 
-    # First call should raise
-    with pytest.raises(RuntimeError, match='Fetch failed'):
+    # First call should raise (wrapped in GenkitError by action tracing)
+    with pytest.raises(GenkitError):
         await dap.list_action_metadata('tool', '*')
     assert call_count == 1
 
@@ -379,15 +382,19 @@ async def dap_fn() -> DapValue:
 
 
 def test_transform_dap_value(tool1: Action, tool2: Action) -> None:
-    """Test the transform_dap_value utility function."""
+    """Test the transform_dap_value utility function.
+
+    Updated for PR #4050 parity: returns flat list instead of grouped dict.
+    """
     value: DapValue = {'tool': [tool1, tool2]}
 
     metadata = transform_dap_value(value)
 
-    assert 'tool' in metadata
-    assert len(metadata['tool']) == 2
-    assert metadata['tool'][0] == tool1.metadata
-    assert metadata['tool'][1] == tool2.metadata
+    # New API returns flat list of action metadata
+    assert isinstance(metadata, list)
+    assert len(metadata) == 2
+    assert metadata[0].get('name') == 'tool1'
+    assert metadata[1].get('name') == 'tool2'
 
 
 def test_dap_config_string_normalization(registry: Registry) -> None:
@@ -531,3 +538,118 @@ async def dap_fn() -> DapValue:
 
     with pytest.raises(ValueError, match='name required'):
         await dap.get_action_metadata_record('dap/my-dap')
+
+
+@pytest.mark.asyncio
+async def test_parse_registry_key_standard_format() -> None:
+    """Test parsing standard registry keys."""
+    # Standard model key
+    parsed = parse_registry_key('/model/googleai/gemini-2.0-flash')
+    assert parsed is not None
+    assert parsed.action_type == 'model'
+    assert parsed.plugin_name == 'googleai'
+    assert parsed.action_name == 'gemini-2.0-flash'
+    assert parsed.dynamic_action_host is None
+
+    # Util key (short format)
+    parsed = parse_registry_key('/util/generate')
+    assert parsed is not None
+    assert parsed.action_type == 'util'
+    assert parsed.action_name == 'generate'
+    assert parsed.plugin_name is None
+
+    # Invalid key
+    parsed = parse_registry_key('invalid')
+    assert parsed is None
+
+
+@pytest.mark.asyncio
+async def test_parse_registry_key_dap_format() -> None:
+    """Test parsing DAP-style registry keys."""
+    # DAP key with action type and name
+    parsed = parse_registry_key('/dynamic-action-provider/mcp-host:tool/my-tool')
+    assert parsed is not None
+    assert parsed.dynamic_action_host == 'mcp-host'
+    assert parsed.action_type == 'tool'
+    assert parsed.action_name == 'my-tool'
+
+    # DAP key without action type (just host)
+    parsed = parse_registry_key('/dynamic-action-provider/mcp-host')
+    assert parsed is not None
+    assert parsed.action_type == 'dynamic-action-provider'
+    assert parsed.action_name == 'mcp-host'
+
+
+@pytest.mark.asyncio
+async def test_list_resolvable_actions_includes_dap(registry: Registry, tool1: Action, tool2: Action) -> None:
+    """Test that list_resolvable_actions includes DAP-provided actions."""
+    call_count = 0
+
+    async def dap_fn() -> DapValue:
+        nonlocal call_count
+        call_count += 1
+        return {'tool': [tool1, tool2]}
+
+    define_dynamic_action_provider(registry, 'test-dap', dap_fn)
+
+    # Get resolvable actions
+    metas = await registry.list_resolvable_actions()
+
+    # Should include the DAP itself and the tools it provides (with full keys)
+    names = [m.name for m in metas]
+    assert 'test-dap' in names  # The DAP action
+    assert '/dynamic-action-provider/test-dap:tool/tool1' in names  # DAP-provided tool
+    assert '/dynamic-action-provider/test-dap:tool/tool2' in names  # DAP-provided tool
+
+
+@pytest.mark.asyncio
+async def test_runs_action_with_transformed_metadata(registry: Registry, tool1: Action, tool2: Action) -> None:
+    """Test that the DAP action returns transformed metadata when run.
+
+    Corresponds to JS test: 'runs the action with transformed metadata when fetching'
+    """
+
+    async def dap_fn() -> DapValue:
+        return {'tool': [tool1, tool2]}
+
+    dap = define_dynamic_action_provider(registry, 'my-dap', dap_fn)
+
+    # Fetch the DAP value through the cache (which runs the action)
+    await dap._cache.get_or_fetch()
+
+    # Run the action directly and check the result is transformed metadata
+    result = await dap.action.arun(None)
+    metadata_list = result.response
+
+    # Should return transformed metadata (list of ActionMetadata-like dicts)
+    assert len(metadata_list) == 2
+    names = [m['name'] for m in metadata_list]
+    assert 'tool1' in names
+    assert 'tool2' in names
+
+
+@pytest.mark.asyncio
+async def test_skips_trace_when_requested(registry: Registry, tool1: Action, tool2: Action) -> None:
+    """Test that skipTrace parameter skips creating a trace.
+
+    Corresponds to JS test: 'skips trace when requested'
+    """
+    call_count = 0
+
+    async def dap_fn() -> DapValue:
+        nonlocal call_count
+        call_count += 1
+        return {'tool': [tool1, tool2]}
+
+    dap = define_dynamic_action_provider(registry, 'my-dap', dap_fn)
+
+    # Fetch with skip_trace=True should call the dap_fn directly (not via action.run)
+    await dap._cache.get_or_fetch(skip_trace=True)
+    assert call_count == 1
+
+    # Invalidate cache
+    dap.invalidate_cache()
+
+    # Fetch without skip_trace should also work (calls via action.run which calls dap_fn)
+    await dap._cache.get_or_fetch(skip_trace=False)
+    assert call_count == 2
diff --git a/py/pyproject.toml b/py/pyproject.toml
index 6d718aef8d..3d733323bd 100644
--- a/py/pyproject.toml
+++ b/py/pyproject.toml
@@ -75,7 +75,7 @@ dev = [
 lint = [
   "bandit>=1.7.0",
   "deptry>=0.22.0",
-  "litestar>=2.0.0",          # For web/typing.py type resolution
+  "litestar>=2.0.0",     # For web/typing.py type resolution
   "mypy>=1.14.0",
   "pip-audit>=2.7.0",
   "pypdf>=6.6.2",
@@ -136,6 +136,7 @@ amazon-bedrock-hello              = { workspace = true }
 anthropic-hello                   = { workspace = true }
 cloudflare-workers-ai-hello       = { workspace = true }
 compat-oai-hello                  = { workspace = true }
+dap-demo                          = { workspace = true }
 deepseek-hello                    = { workspace = true }
 dev-local-vectorstore-hello       = { workspace = true }
 evaluator-demo                    = { workspace = true }
diff --git a/py/samples/dap-demo/README.md b/py/samples/dap-demo/README.md
new file mode 100644
index 0000000000..1e2bd7f2c9
--- /dev/null
+++ b/py/samples/dap-demo/README.md
@@ -0,0 +1,183 @@
+# Dynamic Action Provider (DAP) Demo
+
+This sample demonstrates how to use **Dynamic Action Providers (DAPs)** to
+dynamically provide tools at runtime, enabling integration with external
+systems like MCP servers, plugin registries, or other dynamic tool sources.
+
+## What is a Dynamic Action Provider (DAP)?
+
+A DAP is a factory that creates actions (tools, flows, etc.) at runtime rather
+than at startup. This is useful for:
+
+- **MCP Integration**: Connect to Model Context Protocol servers
+- **Plugin Systems**: Load tools from external plugin registries
+- **Multi-tenant Systems**: Provide tenant-specific tools dynamically
+- **Feature Flags**: Enable/disable tools based on runtime configuration
+
+## Key Concepts
+
+| Concept | Description |
+|---------|-------------|
+| **DAP** | A "tool factory" that creates tools on-demand at runtime |
+| **Dynamic Tool** | A tool created via `ai.dynamic_tool()` - not registered globally |
+| **Cache** | DAP results are cached to avoid recreating tools on every request |
+| **TTL** | Time-To-Live - how long cached tools remain valid before refresh |
+| **Invalidation** | Manually clear the cache to force fresh tool creation |
+
+## Architecture
+
+```
+┌─────────────────────────────────────────────────────────────────────────────┐
+│                         DAP Flow                                             │
+├─────────────────────────────────────────────────────────────────────────────┤
+│                                                                              │
+│  ┌──────────────┐     ┌──────────────┐     ┌──────────────┐                │
+│  │   Genkit     │     │     DAP      │     │   External   │                │
+│  │   generate() │ ──► │    Cache     │ ──► │   System     │                │
+│  └──────────────┘     └──────────────┘     │ (API, DB)    │                │
+│         │                    │              └──────────────┘                │
+│         ▼                    ▼                     │                        │
+│  ┌──────────────┐     ┌──────────────┐             │                        │
+│  │   Model      │ ◄── │   Dynamic    │ ◄───────────┘                        │
+│  │   Response   │     │   Tools      │                                      │
+│  └──────────────┘     └──────────────┘                                      │
+│                                                                              │
+└─────────────────────────────────────────────────────────────────────────────┘
+```
+
+## Features Demonstrated
+
+1. **Multiple DAPs**: Weather tools and Finance tools from separate providers
+2. **Caching Strategies**: Different TTL values for different tool sources
+3. **Cache Invalidation**: Manually refresh tools when needed
+4. **Multi-source Composition**: Combine tools from multiple DAPs in one query
+
+## Prerequisites
+
+- Python 3.10+
+- Google AI API key (`GEMINI_API_KEY`)
+
+## Running the Sample
+
+### Using the Sample Runner (Recommended)
+
+```bash
+# From the py/ directory
+./bin/run_sample dap-demo
+```
+
+### Manual Execution
+
+```bash
+# Navigate to sample directory
+cd py/samples/dap-demo
+
+# Install dependencies
+uv sync
+
+# Run with Genkit DevUI
+./run.sh
+
+# Or run with genkit start directly
+uv run genkit start -- python src/main.py
+```
+
+## Testing in the DevUI
+
+1. **Start the sample** using `./run.sh` or the sample runner
+2. **Open the DevUI** at http://localhost:4000
+3. **Navigate to Flows** in the left sidebar
+4. **Select a flow** (e.g., `weather_assistant`)
+5. **Click Run** - default input values are pre-filled
+6. **View the result** in the response panel
+
+### Testing Each Flow
+
+| Flow | Default Input | Expected Output |
+|------|---------------|-----------------|
+| `weather_assistant` | `{"city": "Tokyo"}` | Weather information for Tokyo |
+| `finance_assistant` | `{"query": "What's the current price of AAPL stock?"}` | List of available finance tools |
+| `multi_assistant` | `{"query": "..."}` | List of all tools from both DAPs |
+| `refresh_tools_demo` | `{"source": "all"}` | Cache invalidation confirmation |
+| `list_dap_tools` | `{"source": "all"}` | List of all available tool names |
+
+## Available Flows
+
+### weather_assistant
+
+Get weather information for a city using the dynamically-provided weather tool.
+
+**Input**: `WeatherInput` with `city` field (default: "Tokyo")
+
+### finance_assistant
+
+Answer finance questions using dynamically-provided finance tools.
+
+**Input**: `FinanceInput` with `query` field (default: "What's the current price of AAPL stock?")
+
+### multi_assistant
+
+Multi-source assistant that combines tools from both Weather and Finance DAPs.
+
+**Input**: `MultiInput` with `query` field
+
+### refresh_tools_demo
+
+Invalidate DAP cache to force fresh tool fetching.
+
+**Input**: `RefreshInput` with `source` field ("weather", "finance", or "all")
+
+### list_dap_tools
+
+List all tools provided by a specific DAP or all DAPs.
+
+**Input**: `ListToolsInput` with `source` field ("weather", "finance", or "all")
+
+## DAP Configuration Examples
+
+### Weather Tools DAP (Short Cache)
+
+```python
+weather_dap = ai.define_dynamic_action_provider(
+    config=DapConfig(
+        name='weather-tools',
+        description='Provides weather-related tools',
+        cache_config=DapCacheConfig(ttl_millis=5000),  # 5 second cache
+    ),
+    fn=weather_tools_provider,
+)
+```
+
+### Finance Tools DAP (Long Cache)
+
+```python
+finance_dap = ai.define_dynamic_action_provider(
+    config=DapConfig(
+        name='finance-tools',
+        description='Provides finance and market tools',
+        cache_config=DapCacheConfig(ttl_millis=60000),  # 60 second cache
+    ),
+    fn=finance_tools_provider,
+)
+```
+
+## Use Cases
+
+1. **MCP Integration**: Use DAPs to connect to MCP servers and expose their
+   tools to Genkit. See the `genkit-plugin-mcp` package for a complete
+   implementation.
+
+2. **Plugin Marketplace**: Load tools from an external registry based on
+   user preferences or subscription level.
+
+3. **Multi-tenant SaaS**: Provide different tools to different tenants based
+   on their configuration or tier.
+
+4. **A/B Testing**: Enable different tool sets for different users to test
+   effectiveness.
+
+## Related Resources
+
+- [Genkit Python SDK Documentation](https://firebase.google.com/docs/genkit/python)
+- [MCP Plugin](../../plugins/mcp/) - Full MCP integration using DAP
+- [JS DAP Implementation](../../../../js/core/src/dynamic-action-provider.ts)
diff --git a/py/samples/dap-demo/pyproject.toml b/py/samples/dap-demo/pyproject.toml
new file mode 100644
index 0000000000..f6b04188ad
--- /dev/null
+++ b/py/samples/dap-demo/pyproject.toml
@@ -0,0 +1,59 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# SPDX-License-Identifier: Apache-2.0
+
+[project]
+authors = [{ name = "Google" }]
+classifiers = [
+  "Development Status :: 3 - Alpha",
+  "Environment :: Console",
+  "Environment :: Web Environment",
+  "Intended Audience :: Developers",
+  "Operating System :: OS Independent",
+  "Programming Language :: Python",
+  "Programming Language :: Python :: 3 :: Only",
+  "Programming Language :: Python :: 3.10",
+  "Programming Language :: Python :: 3.11",
+  "Programming Language :: Python :: 3.12",
+  "Programming Language :: Python :: 3.13",
+  "Programming Language :: Python :: 3.14",
+  "Topic :: Scientific/Engineering :: Artificial Intelligence",
+  "Topic :: Software Development :: Libraries",
+]
+dependencies = [
+  "genkit",
+  "genkit-plugin-google-genai",
+  "pydantic>=2.10.5",
+  "rich>=13.0.0",
+]
+description = "Dynamic Action Provider (DAP) demonstration sample"
+license = "Apache-2.0"
+name = "dap-demo"
+readme = "README.md"
+requires-python = ">=3.10"
+version = "0.1.0"
+
+[project.optional-dependencies]
+dev = ["watchdog>=6.0.0"]
+
+[build-system]
+build-backend = "hatchling.build"
+requires      = ["hatchling"]
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/dap_demo"]
+
+[tool.hatch.metadata]
+allow-direct-references = true
diff --git a/py/samples/dap-demo/run.sh b/py/samples/dap-demo/run.sh
new file mode 100755
index 0000000000..f6fc6677d6
--- /dev/null
+++ b/py/samples/dap-demo/run.sh
@@ -0,0 +1,59 @@
+#!/usr/bin/env bash
+# Copyright 2026 Google LLC
+# SPDX-License-Identifier: Apache-2.0
+
+# Dynamic Action Provider (DAP) Demo
+# ===================================
+#
+# Demonstrates how to use Dynamic Action Providers to provide tools at runtime.
+#
+# Prerequisites:
+#   - GEMINI_API_KEY environment variable set
+#
+# Usage:
+#   ./run.sh          # Start the demo with Dev UI
+#   ./run.sh --help   # Show this help message
+
+set -euo pipefail
+
+cd "$(dirname "$0")"
+source "../_common.sh"
+
+print_help() {
+    print_banner "DAP Demo" "🔌"
+    echo "Usage: ./run.sh [options]"
+    echo ""
+    echo "Options:"
+    echo "  --help     Show this help message"
+    echo ""
+    echo "Environment Variables:"
+    echo "  GEMINI_API_KEY    Required. Your Gemini API key"
+    echo ""
+    echo "Get an API key from: https://makersuite.google.com/app/apikey"
+    print_help_footer
+}
+
+# Parse arguments
+case "${1:-}" in
+    --help|-h)
+        print_help
+        exit 0
+        ;;
+esac
+
+# Main execution
+print_banner "DAP Demo" "🔌"
+
+check_env_var "GEMINI_API_KEY" "https://makersuite.google.com/app/apikey" || true
+
+install_deps
+
+# Start with hot reloading and auto-open browser
+genkit_start_with_browser -- \
+    uv tool run --from watchdog watchmedo auto-restart \
+        -d src \
+        -d ../../packages \
+        -d ../../plugins \
+        -p '*.py;*.prompt;*.json' \
+        -R \
+        -- uv run src/dap_demo/__init__.py "$@"
diff --git a/py/samples/dap-demo/src/dap_demo/__init__.py b/py/samples/dap-demo/src/dap_demo/__init__.py
new file mode 100644
index 0000000000..892f4cb622
--- /dev/null
+++ b/py/samples/dap-demo/src/dap_demo/__init__.py
@@ -0,0 +1,514 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# SPDX-License-Identifier: Apache-2.0
+
+"""Dynamic Action Provider (DAP) Demo.
+
+This sample demonstrates how to use Dynamic Action Providers (DAPs) to provide
+tools dynamically at runtime. DAPs are useful for integrating external tool
+sources like MCP servers, plugin registries, or service meshes.
+
+What is DAP? (ELI5 - The Toy Box Analogy)
+-----------------------------------------
+
+Imagine you have two ways to get toys:
+
+**Regular Tools (Static)** - Your Toy Box at Home::
+
+    📦 Your Toy Box
+    ├── 🚗 Car (always there)
+    ├── 🧸 Teddy Bear (always there)
+    └── 🎮 Game (always there)
+
+You know exactly what toys you have. They're always in the same spot.
+This is like regular ``@ai.tool()`` - defined once at startup, always available.
+
+**DAP Tools (Dynamic)** - A Toy Rental Store::
+
+    🏪 Toy Rental Store (DAP)
+    ├── "What toys do you have today?"
+    ├── Store checks inventory...
+    └── "Today we have: 🚀 Rocket, 🦖 Dinosaur, 🎸 Guitar!"
+
+The toys change! You ask the store, they check what's in stock RIGHT NOW,
+and give you options. Tomorrow might be different toys!
+
+Static vs Dynamic Tools::
+
+    ┌─────────────────────────────────────────────────────────────────┐
+    │                        WITHOUT DAP                               │
+    │                                                                  │
+    │   Your App Starts                                                │
+    │        │                                                         │
+    │        ▼                                                         │
+    │   ┌─────────────┐                                                │
+    │   │ Define tool │  @ai.tool("get_weather")                      │
+    │   │ Define tool │  @ai.tool("get_stocks")                       │
+    │   └─────────────┘                                                │
+    │        │                                                         │
+    │        ▼                                                         │
+    │   Tools are FIXED. Can't add new ones without restarting.       │
+    └─────────────────────────────────────────────────────────────────┘
+
+    ┌─────────────────────────────────────────────────────────────────┐
+    │                         WITH DAP                                 │
+    │                                                                  │
+    │   Your App Starts                                                │
+    │        │                                                         │
+    │        ▼                                                         │
+    │   ┌───────────────────┐                                          │
+    │   │ Register DAP      │  "Ask MCP server for tools"             │
+    │   └───────────────────┘                                          │
+    │        │                                                         │
+    │        ▼                                                         │
+    │   When you need tools...                                         │
+    │        │                                                         │
+    │        ▼                                                         │
+    │   ┌───────────────────┐     ┌─────────────────────┐             │
+    │   │ DAP asks server   │ ──► │ MCP Server says:    │             │
+    │   │ "What tools now?" │     │ "I have 5 tools!"   │             │
+    │   └───────────────────┘     └─────────────────────┘             │
+    │        │                                                         │
+    │        ▼                                                         │
+    │   Tools could be DIFFERENT each time! 🎉                        │
+    └─────────────────────────────────────────────────────────────────┘
+
+Key Concepts::
+
+    ┌─────────────────────┬────────────────────────────────────────────────┐
+    │ Concept             │ ELI5 Explanation                               │
+    ├─────────────────────┼────────────────────────────────────────────────┤
+    │ DAP                 │ A "tool factory" that creates tools on-demand. │
+    │                     │ Like asking a store what's in stock.           │
+    ├─────────────────────┼────────────────────────────────────────────────┤
+    │ Dynamic Tool        │ A tool created at runtime, not at startup.     │
+    │                     │ Like ordering custom pizza vs frozen.          │
+    ├─────────────────────┼────────────────────────────────────────────────┤
+    │ Cache               │ Remembers tools to avoid recreating them.      │
+    │                     │ Like a notepad to avoid asking twice.          │
+    ├─────────────────────┼────────────────────────────────────────────────┤
+    │ TTL (Time-To-Live)  │ How long cached tools stay fresh.              │
+    │                     │ Like an expiration date on milk.               │
+    ├─────────────────────┼────────────────────────────────────────────────┤
+    │ Invalidation        │ Throwing away stale cached tools.              │
+    │                     │ Like clearing your browser cache.              │
+    └─────────────────────┴────────────────────────────────────────────────┘
+
+Why Use DAP?
+------------
+
+1. **MCP Servers** - Connect to external tool servers that add/remove tools
+2. **Plugin Systems** - Users can install new tools without restarting
+3. **Multi-tenant** - Different users might have access to different tools
+4. **Service Mesh** - Tools discovered from a network of microservices
+
+Data Flow::
+
+    User Request
+         │
+         ▼
+    ┌─────────────────┐
+    │     Flow        │  (e.g., weather_assistant)
+    └────────┬────────┘
+             │
+             ▼
+    ┌─────────────────┐     ┌─────────────────┐
+    │      DAP        │ ──► │   Tool Cache    │
+    │  (weather-tools)│     │   (TTL: 5s)     │
+    └────────┬────────┘     └─────────────────┘
+             │
+             │ Cache miss? Call provider function
+             ▼
+    ┌─────────────────┐
+    │  Tool Provider  │  Creates dynamic tools
+    │    Function     │  (get_weather, etc.)
+    └────────┬────────┘
+             │
+             ▼
+    ┌─────────────────┐
+    │  Dynamic Tool   │  Used in AI generation
+    │   Execution     │
+    └────────┬────────┘
+             │
+             ▼
+    ┌─────────────────┐
+    │    Response     │
+    └─────────────────┘
+
+Running the Demo
+----------------
+
+1. Navigate to the sample directory::
+
+    cd py/samples/dap-demo
+
+2. Run with the sample runner::
+
+    ../../bin/run_sample dap-demo
+
+3. Or run directly with genkit start::
+
+    uv run genkit start -- python src/dap_demo/__init__.py
+
+Testing in the DevUI
+--------------------
+
+1. Open http://localhost:4000
+2. Navigate to Flows
+3. Select any flow (e.g., 'weather_assistant')
+4. The default input values are pre-filled - just click Run
+
+Available Flows
+---------------
+
+- **weather_assistant**: Get weather for a city using dynamic tools
+- **finance_assistant**: Answer finance questions with multiple tools
+- **multi_assistant**: Combine tools from multiple DAPs
+- **refresh_tools_demo**: Demonstrate cache invalidation
+- **list_dap_tools**: List all available tools from DAPs
+"""
+
+import asyncio
+import random
+import re
+from typing import Any
+
+from pydantic import BaseModel, Field
+from rich.traceback import install as install_rich_traceback
+
+from genkit.ai import Genkit
+from genkit.blocks.dap import DapCacheConfig, DapConfig, DapValue
+from genkit.core.action import Action
+from genkit.plugins.google_genai import GoogleAI
+
+install_rich_traceback(show_locals=True, width=120, extra_lines=3)
+
+ai = Genkit(
+    plugins=[GoogleAI()],
+)
+
+
+class WeatherInput(BaseModel):
+    """Input for weather assistant flow."""
+
+    city: str = Field(default='Tokyo', description='City name to get weather for')
+
+
+class FinanceInput(BaseModel):
+    """Input for finance assistant flow."""
+
+    query: str = Field(default="What's the current price of AAPL stock?", description='Finance question to answer')
+
+
+class MultiInput(BaseModel):
+    """Input for multi-source assistant flow."""
+
+    query: str = Field(
+        default="What's the weather in London and how is the EUR/USD exchange rate?",
+        description='Question that may require multiple tool sources',
+    )
+
+
+class RefreshInput(BaseModel):
+    """Input for cache refresh demo flow."""
+
+    source: str = Field(default='all', description="Which DAP to refresh: 'weather', 'finance', or 'all'")
+
+
+class ListToolsInput(BaseModel):
+    """Input for list tools flow."""
+
+    source: str = Field(default='all', description="Which DAP to list: 'weather', 'finance', or 'all'")
+
+
+class CurrencyConversion(BaseModel):
+    """Input for currency conversion tool."""
+
+    amount: float = Field(default=100.0, description='Amount to convert')
+    from_currency: str = Field(default='USD', description='Source currency code')
+    to_currency: str = Field(default='EUR', description='Target currency code')
+
+
+async def fetch_weather_data(city: str) -> dict[str, Any]:
+    """Simulate fetching weather from an external API."""
+    await asyncio.sleep(0.1)  # Simulate network delay
+    return {
+        'city': city,
+        'temperature': random.randint(15, 35),
+        'conditions': random.choice(['Sunny', 'Cloudy', 'Rainy', 'Windy']),
+        'humidity': random.randint(30, 80),
+    }
+
+
+async def fetch_stock_price(symbol: str) -> dict[str, Any]:
+    """Simulate fetching stock price from an external API."""
+    await asyncio.sleep(0.1)  # Simulate network delay
+    return {
+        'symbol': symbol.upper(),
+        'price': round(random.uniform(50, 500), 2),
+        'change': round(random.uniform(-5, 5), 2),
+        'volume': random.randint(1000000, 10000000),
+    }
+
+
+async def weather_tools_provider() -> DapValue:
+    """DAP function that provides weather-related tools.
+
+    In a real scenario, this could connect to an MCP server, load tools from
+    a plugin registry, or discover tools from a service mesh.
+
+    Uses ai.dynamic_tool() to create unregistered tools that are returned
+    directly to consumers without being in the global registry.
+    """
+
+    async def get_weather_impl(city: str) -> str:
+        """Get current weather for a city."""
+        data = await fetch_weather_data(city)
+        return (
+            f'Weather in {data["city"]}: {data["temperature"]}°C, {data["conditions"]}, Humidity: {data["humidity"]}%'
+        )
+
+    get_weather = ai.dynamic_tool(
+        name='get_weather',
+        fn=get_weather_impl,
+        description='Get current weather for a city',
+    )
+
+    return {'tool': [get_weather]}
+
+
+async def finance_tools_provider() -> DapValue:
+    """DAP function that provides finance-related tools.
+
+    This provider has a longer cache TTL because the available tools change
+    less frequently than weather tools.
+    """
+
+    async def get_stock_price_impl(symbol: str) -> str:
+        """Get current stock price by symbol."""
+        data = await fetch_stock_price(symbol)
+        change_str = f'+{data["change"]}' if data['change'] > 0 else str(data['change'])
+        return f'{data["symbol"]}: ${data["price"]} ({change_str}%), Volume: {data["volume"]:,}'
+
+    async def convert_currency_impl(input: CurrencyConversion) -> str:
+        """Convert between currencies."""
+        rates = {'USD': 1.0, 'EUR': 0.85, 'GBP': 0.73, 'JPY': 110.0}
+        from_rate = rates.get(input.from_currency.upper(), 1.0)
+        to_rate = rates.get(input.to_currency.upper(), 1.0)
+        converted = input.amount / from_rate * to_rate
+        return f'{input.amount} {input.from_currency.upper()} = {converted:.2f} {input.to_currency.upper()}'
+
+    get_stock_price = ai.dynamic_tool(
+        name='get_stock_price',
+        fn=get_stock_price_impl,
+        description='Get current stock price by symbol',
+    )
+
+    convert_currency = ai.dynamic_tool(
+        name='convert_currency',
+        fn=convert_currency_impl,
+        description='Convert between currencies',
+    )
+
+    return {'tool': [get_stock_price, convert_currency]}
+
+
+weather_dap = ai.define_dynamic_action_provider(
+    config=DapConfig(
+        name='weather-tools',
+        description='Provides weather-related tools',
+        cache_config=DapCacheConfig(ttl_millis=5000),
+    ),
+    fn=weather_tools_provider,
+)
+
+finance_dap = ai.define_dynamic_action_provider(
+    config=DapConfig(
+        name='finance-tools',
+        description='Provides finance and market tools',
+        cache_config=DapCacheConfig(ttl_millis=60000),
+    ),
+    fn=finance_tools_provider,
+)
+
+
+@ai.flow(description='Weather assistant using dynamically-provided tools')
+async def weather_assistant(input: WeatherInput) -> str:
+    """Get weather information for a city using dynamic tools.
+
+    The weather tool is provided by the weather-tools DAP, which could be
+    sourced from an MCP server, plugin registry, or other external system.
+    """
+    weather_tool = await weather_dap.get_action('tool', 'get_weather')
+
+    if not weather_tool:
+        return f'Weather service unavailable. Cannot get weather for {input.city}.'
+
+    result = await weather_tool.arun(input.city)
+    return str(result.response)
+
+
+@ai.flow(description='Finance assistant using dynamically-provided tools')
+async def finance_assistant(input: FinanceInput) -> str:
+    """Answer finance questions using dynamic tools.
+
+    The finance tools are provided by the finance-tools DAP with a longer
+    cache TTL since the available tools change less frequently.
+
+    This flow demonstrates using a model to answer queries with dynamic tools.
+    """
+    cache_result = await finance_dap._cache.get_or_fetch()  # noqa: SLF001 - accessing internal cache for demo
+    tools = cache_result.get('tool', [])
+
+    if not tools:
+        return 'Finance service unavailable.'
+
+    # Use a model to answer the query using the dynamic tools
+    # Note: Dynamic tools are not in the global registry, so we invoke them directly.
+    # For stock queries, use the get_stock_price tool
+    get_stock_price = next((t for t in tools if t.name == 'get_stock_price'), None)
+    if get_stock_price and 'stock' in input.query.lower():
+        # Extract stock symbol from query (clean punctuation, search from end)
+        cleaned_words = (re.sub(r'[^A-Z0-9]', '', w) for w in reversed(input.query.upper().split()))
+        symbol = next((w for w in cleaned_words if w and 1 <= len(w) <= 5), 'AAPL')
+        result = await get_stock_price.arun(symbol)
+        return str(result.response)
+
+    # For currency queries, use the convert_currency tool
+    convert_currency = next((t for t in tools if t.name == 'convert_currency'), None)
+    if convert_currency and any(word in input.query.lower() for word in ['convert', 'exchange', 'currency']):
+        result = await convert_currency.arun(CurrencyConversion())
+        return str(result.response)
+
+    tool_names = [t.name for t in tools]
+    return f'Available finance tools: {", ".join(tool_names)}. Try asking about stocks or currency conversion.'
+
+
+@ai.flow(description='Multi-source assistant combining tools from multiple DAPs')
+async def multi_assistant(input: MultiInput) -> str:
+    """Assistant that can use tools from multiple DAPs.
+
+    This demonstrates how DAPs can be composed to provide tools from
+    multiple sources (weather service + finance APIs) in a single query.
+
+    Uses asyncio.gather for concurrent fetching.
+    """
+    all_tools: list[Action[Any, Any]] = []
+
+    # Fetch from both DAPs concurrently for efficiency
+    weather_cache, finance_cache = await asyncio.gather(
+        weather_dap._cache.get_or_fetch(),  # noqa: SLF001
+        finance_dap._cache.get_or_fetch(),  # noqa: SLF001
+    )
+    all_tools.extend(weather_cache.get('tool', []))
+    all_tools.extend(finance_cache.get('tool', []))
+
+    if not all_tools:
+        return 'No tools available.'
+
+    # Demonstrate composing tools from multiple sources
+    # Collect results from all matching tools
+    results: list[str] = []
+
+    # For weather queries, use the weather tool
+    get_weather = next((t for t in all_tools if t.name == 'get_weather'), None)
+    if get_weather and 'weather' in input.query.lower():
+        # Extract city name (simple heuristic - use first capitalized word after 'in')
+        match = re.search(r'\bin\s+(\w+)', input.query, re.IGNORECASE)
+        city = match.group(1) if match else 'London'
+        result = await get_weather.arun(city)
+        results.append(str(result.response))
+
+    # For currency/exchange queries, use convert_currency tool
+    convert_currency = next((t for t in all_tools if t.name == 'convert_currency'), None)
+    if convert_currency and any(word in input.query.lower() for word in ['eur', 'usd', 'exchange', 'currency', 'rate']):
+        result = await convert_currency.arun(CurrencyConversion(from_currency='EUR', to_currency='USD'))
+        results.append(str(result.response))
+
+    # For stock queries, use get_stock_price tool
+    get_stock_price = next((t for t in all_tools if t.name == 'get_stock_price'), None)
+    if get_stock_price and any(word in input.query.lower() for word in ['stock', 'price', 'aapl']):
+        result = await get_stock_price.arun('AAPL')
+        results.append(str(result.response))
+
+    if results:
+        return ' | '.join(results)
+
+    tool_names = [t.name for t in all_tools]
+    return f'Available tools: {", ".join(tool_names)}. Try asking about weather, stocks, or currency.'
+
+
+@ai.flow(description='Demonstrate DAP cache invalidation')
+async def refresh_tools_demo(input: RefreshInput) -> str:
+    """Invalidate and refresh dynamic tools.
+
+    In a real scenario, you might invalidate the cache when:
+    - An MCP server restarts
+    - A plugin is added/removed
+    - Configuration changes
+    """
+    if input.source == 'weather':
+        weather_dap.invalidate_cache()
+        return 'Weather tools cache invalidated. Next request will fetch fresh tools.'
+    elif input.source == 'finance':
+        finance_dap.invalidate_cache()
+        return 'Finance tools cache invalidated. Next request will fetch fresh tools.'
+    elif input.source == 'all':
+        weather_dap.invalidate_cache()
+        finance_dap.invalidate_cache()
+        return 'All DAP caches invalidated. Next requests will fetch fresh tools.'
+    else:
+        return f"Unknown source: {input.source}. Use 'weather', 'finance', or 'all'."
+
+
+@ai.flow(description='List all tools available from DAPs')
+async def list_dap_tools(input: ListToolsInput) -> str:
+    """List all tools provided by a specific DAP or all DAPs.
+
+    This demonstrates retrieving tools from DAP cache.
+    """
+    if input.source == 'weather':
+        cache = await weather_dap._cache.get_or_fetch()  # noqa: SLF001
+        tools = cache.get('tool', [])
+        return f'Weather tools: {[t.name for t in tools]}'
+    elif input.source == 'finance':
+        cache = await finance_dap._cache.get_or_fetch()  # noqa: SLF001
+        tools = cache.get('tool', [])
+        return f'Finance tools: {[t.name for t in tools]}'
+    elif input.source == 'all':
+        # Fetch from both DAPs concurrently for efficiency
+        weather_cache, finance_cache = await asyncio.gather(
+            weather_dap._cache.get_or_fetch(),  # noqa: SLF001
+            finance_dap._cache.get_or_fetch(),  # noqa: SLF001
+        )
+        weather_tools = weather_cache.get('tool', [])
+        finance_tools = finance_cache.get('tool', [])
+        all_names = [t.name for t in weather_tools] + [t.name for t in finance_tools]
+        return f'All available tools: {all_names}'
+    else:
+        return f"Unknown source: {input.source}. Use 'weather', 'finance', or 'all'."
+
+
+async def main() -> None:
+    """Keep the server running for the DevUI.
+
+    When running with 'genkit start', this keeps the process alive so flows
+    can be tested through the DevUI at http://localhost:4000.
+    """
+    await asyncio.Event().wait()
+
+
+if __name__ == '__main__':
+    ai.run_main(main())
diff --git a/py/uv.lock b/py/uv.lock
index 0d84d2ae2b..6b35075e71 100644
--- a/py/uv.lock
+++ b/py/uv.lock
@@ -15,6 +15,7 @@ members = [
     "anthropic-hello",
     "cloudflare-workers-ai-hello",
     "compat-oai-hello",
+    "dap-demo",
     "deepseek-hello",
     "dev-local-vectorstore-hello",
     "evaluator-demo",
@@ -1319,6 +1320,32 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/c7/1b/534ad8a5e0f9470522811a8e5a9bc5d328fb7738ba29faf357467a4ef6d0/cyclonedx_python_lib-11.6.0-py3-none-any.whl", hash = "sha256:94f4aae97db42a452134dafdddcfab9745324198201c4777ed131e64c8380759", size = 511157, upload-time = "2025-12-02T12:28:44.158Z" },
 ]
 
+[[package]]
+name = "dap-demo"
+version = "0.1.0"
+source = { editable = "samples/dap-demo" }
+dependencies = [
+    { name = "genkit" },
+    { name = "genkit-plugin-google-genai" },
+    { name = "pydantic" },
+    { name = "rich" },
+]
+
+[package.optional-dependencies]
+dev = [
+    { name = "watchdog" },
+]
+
+[package.metadata]
+requires-dist = [
+    { name = "genkit", editable = "packages/genkit" },
+    { name = "genkit-plugin-google-genai", editable = "plugins/google-genai" },
+    { name = "pydantic", specifier = ">=2.10.5" },
+    { name = "rich", specifier = ">=13.0.0" },
+    { name = "watchdog", marker = "extra == 'dev'", specifier = ">=6.0.0" },
+]
+provides-extras = ["dev"]
+
 [[package]]
 name = "datamodel-code-generator"
 version = "0.53.0"