Skip to content

Backend Abstraction Layer

The backend abstraction layer provides a provider-agnostic interface for agent execution. Downstream consumers (test runner, chat session, serve endpoint) depend only on the protocols defined in holodeck.lib.backends.base -- no provider-specific types leak through.

Routing

BackendSelector inspects model.provider and instantiates the correct backend automatically:

Provider Backend
openai, azure_openai, ollama SKBackend
anthropic ClaudeBackend

holodeck.lib.backends.base -- Core Protocols & Data Classes

Defines the provider-agnostic contracts that every backend must satisfy and the unified result types returned to callers.

ExecutionResult

ExecutionResult(response, tool_calls=list(), tool_results=list(), token_usage=TokenUsage.zero(), structured_output=None, num_turns=1, is_error=False, error_reason=None) dataclass

Provider-agnostic result of a single agent turn.

Attributes:

Name Type Description
response str

The text response from the agent.
tool_calls list[dict[str, Any]]

List of tool call records made during execution.
tool_results list[dict[str, Any]]

List of tool result records returned during execution.
token_usage TokenUsage

Token consumption metadata for this turn.
structured_output Any | None

Optional structured output from the agent.
num_turns int

Number of turns taken to produce this result.
is_error bool

Whether the execution ended in an error state.
error_reason str | None

Human-readable reason for the error, if any.

ToolEvent

ToolEvent(kind, tool_name, tool_use_id, tool_input=None, tool_response=None, error=None) dataclass

Real-time tool execution event from the backend.

Emitted by backends that support hook-based tool observation (e.g. Claude Agent SDK). Events are pushed onto an asyncio.Queue that consumers can drain concurrently during agent execution.

Attributes:

Name Type Description
kind Literal['start', 'end', 'error']

Event type — "start" before execution, "end" after success, "error" after failure.
tool_name str

Name of the tool being invoked.
tool_use_id str

Unique identifier correlating start/end/error for the same invocation.
tool_input dict[str, Any] | None

Tool input parameters (present on "start").
tool_response str | None

Tool output (present on "end").
error str | None

Error description (present on "error").

AgentSession

AgentSession

Bases: Protocol

Stateful multi-turn conversation session.

Implementations maintain conversation history across multiple send calls. Callers must invoke close when the session is no longer needed to release any held resources (connections, subprocesses, etc.).

close() async

Release session resources (connections, subprocesses, etc.).

Source code in src/holodeck/lib/backends/base.py 
106
107
108
async def close(self) -> None:
    """Release session resources (connections, subprocesses, etc.)."""
    ...

send(message) async

Send a message and receive a single-turn result.

Parameters:

Name Type Description Default
message str

The user message to send to the agent.

 required

Returns:

Type Description
ExecutionResult

ExecutionResult containing the agent response and metadata.
Source code in src/holodeck/lib/backends/base.py 
83
84
85
86
87
88
89
90
91
92
async def send(self, message: str) -> ExecutionResult:
    """Send a message and receive a single-turn result.

    Args:
        message: The user message to send to the agent.

    Returns:
        ExecutionResult containing the agent response and metadata.
    """
    ...

send_streaming(message) async

Send a message and stream the agent response token by token.

Parameters:

Name Type Description Default
message str

The user message to send to the agent.

 required

Yields:

Type Description
AsyncGenerator[str, None]

Successive string chunks of the agent response.
Source code in src/holodeck/lib/backends/base.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
async def send_streaming(self, message: str) -> AsyncGenerator[str, None]:
    """Send a message and stream the agent response token by token.

    Args:
        message: The user message to send to the agent.

    Yields:
        Successive string chunks of the agent response.
    """
    # Protocol stub — concrete implementations use `yield`
    yield ""  # pragma: no cover

AgentBackend

AgentBackend

Bases: Protocol

Provider backend factory.

Each backend encapsulates provider-specific initialisation logic and exposes a uniform surface for single-turn invocations (invoke_once) and stateful sessions (create_session). Callers must call initialize before any other method and teardown when done.

create_session() async

Create a new stateful multi-turn session.

Returns:

Type Description
AgentSession

A fresh AgentSession instance bound to this backend.

Raises:

Type Description
BackendInitError

If the backend was not initialised before calling.
BackendSessionError

If the session cannot be created.
Source code in src/holodeck/lib/backends/base.py 
150
151
152
153
154
155
156
157
158
159
160
async def create_session(self) -> AgentSession:
    """Create a new stateful multi-turn session.

    Returns:
        A fresh AgentSession instance bound to this backend.

    Raises:
        BackendInitError: If the backend was not initialised before calling.
        BackendSessionError: If the session cannot be created.
    """
    ...

initialize() async

Prepare the backend for use.

Raises:

Type Description
BackendInitError

If the backend cannot be initialised (e.g. missing API key, unavailable subprocess).
Source code in src/holodeck/lib/backends/base.py 
121
122
123
124
125
126
127
128
async def initialize(self) -> None:
    """Prepare the backend for use.

    Raises:
        BackendInitError: If the backend cannot be initialised (e.g.
            missing API key, unavailable subprocess).
    """
    ...

invoke_once(message, context=None) async

Execute a single stateless agent turn.

Parameters:

Name Type Description Default
message str

The user message to send to the agent.

 required
context list[dict[str, Any]] | None

Optional list of prior conversation turns.

 None

Returns:

Type Description
ExecutionResult

ExecutionResult containing the agent response and metadata.

Raises:

Type Description
BackendSessionError

If the invocation fails at runtime.
BackendTimeoutError

If the invocation exceeds configured timeout.
Source code in src/holodeck/lib/backends/base.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
async def invoke_once(
    self,
    message: str,
    context: list[dict[str, Any]] | None = None,
) -> ExecutionResult:
    """Execute a single stateless agent turn.

    Args:
        message: The user message to send to the agent.
        context: Optional list of prior conversation turns.

    Returns:
        ExecutionResult containing the agent response and metadata.

    Raises:
        BackendSessionError: If the invocation fails at runtime.
        BackendTimeoutError: If the invocation exceeds configured timeout.
    """
    ...

teardown() async

Release all backend resources.

Source code in src/holodeck/lib/backends/base.py 
162
163
164
async def teardown(self) -> None:
    """Release all backend resources."""
    ...

ContextGenerator

ContextGenerator

Bases: Protocol

Backend-agnostic contextual embedding generation.

Implementations produce situating context for document chunks by summarising each chunk's role within the larger document. Both the existing Semantic Kernel generator and future Claude SDK generator should satisfy this protocol.

contextualize_batch(chunks, document_text, concurrency=None) async

Generate contextual descriptions for a batch of chunks.

Parameters:

Name Type Description Default
chunks list[DocumentChunk]

Document chunks to contextualize.

 required
document_text str

Full text of the source document.

 required
concurrency int | None

Maximum number of concurrent LLM calls.

 None

Returns:

Type Description
list[str]

A list of contextual description strings, one per chunk.
Source code in src/holodeck/lib/backends/base.py 
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
async def contextualize_batch(
    self,
    chunks: list[DocumentChunk],
    document_text: str,
    concurrency: int | None = None,
) -> list[str]:
    """Generate contextual descriptions for a batch of chunks.

    Args:
        chunks: Document chunks to contextualize.
        document_text: Full text of the source document.
        concurrency: Maximum number of concurrent LLM calls.

    Returns:
        A list of contextual description strings, one per chunk.
    """
    ...

Exceptions

BackendError

Bases: HoloDeckError

Base exception for all backend errors.

Catch this to handle any backend-related failure without needing to know the specific subtype.

BackendInitError

Bases: BackendError

Raised during initialize() — startup validation failures.

Examples include a missing API key, an unreachable subprocess, or an incompatible runtime environment.

BackendSessionError

Bases: BackendError

Raised during send() — session-level failures.

Examples include unexpected disconnections, malformed responses, or provider-reported errors during an active session.

BackendTimeoutError

Bases: BackendError

Raised when a single invocation exceeds the configured timeout.

Callers may choose to retry with a longer timeout or surface this as a user-visible error.

holodeck.lib.backends.selector -- Backend Routing

Routes an Agent configuration to the correct backend based on model.provider.

BackendSelector

BackendSelector

Selects and initializes the appropriate backend for an agent configuration.

select(agent, tool_instances=None, mode='test', allow_side_effects=False) async staticmethod

Select and initialize the appropriate backend for the given agent.

Parameters:

Name Type Description Default
agent Agent

Agent configuration with model provider information.

 required
tool_instances dict[str, Any] | None

Initialized tool instances for Claude backend.

 None
mode str

Execution mode ("test" or "chat").

 'test'
allow_side_effects bool

Allow bash/file_system.write in test mode.

 False

Returns:

Type Description
AgentBackend

An initialized AgentBackend instance ready for use.

Raises:

Type Description
BackendInitError

If the provider is not supported or initialization fails.
Source code in src/holodeck/lib/backends/selector.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
@staticmethod
async def select(
    agent: Agent,
    tool_instances: dict[str, Any] | None = None,
    mode: str = "test",
    allow_side_effects: bool = False,
) -> AgentBackend:
    """Select and initialize the appropriate backend for the given agent.

    Args:
        agent: Agent configuration with model provider information.
        tool_instances: Initialized tool instances for Claude backend.
        mode: Execution mode (``"test"`` or ``"chat"``).
        allow_side_effects: Allow bash/file_system.write in test mode.

    Returns:
        An initialized AgentBackend instance ready for use.

    Raises:
        BackendInitError: If the provider is not supported or initialization fails.
    """
    provider = agent.model.provider

    if provider in (
        ProviderEnum.OPENAI,
        ProviderEnum.AZURE_OPENAI,
        ProviderEnum.OLLAMA,
    ):
        backend = SKBackend(agent_config=agent)
        await backend.initialize()
        return backend

    if provider == ProviderEnum.ANTHROPIC:
        claude_backend = ClaudeBackend(
            agent=agent,
            tool_instances=tool_instances,
            mode=mode,
            allow_side_effects=allow_side_effects,
        )
        await claude_backend.initialize()
        return claude_backend

    raise BackendInitError(f"Unsupported provider: {provider}")

holodeck.lib.backends.sk_backend -- Semantic Kernel Backend

Wraps the existing AgentFactory / AgentThreadRun infrastructure behind the provider-agnostic backend interfaces. Handles OpenAI, Azure OpenAI, and Ollama providers.

SKBackend

SKBackend(agent_config)

Semantic Kernel backend implementing the AgentBackend protocol.

Wraps AgentFactory to provide the provider-agnostic backend interface used by downstream consumers.

Initialize the SK backend with agent configuration.

Parameters:

Name Type Description Default
agent_config Agent

Agent configuration with model and instructions.

 required
Source code in src/holodeck/lib/backends/sk_backend.py 
103
104
105
106
107
108
109
def __init__(self, agent_config: Agent) -> None:
    """Initialize the SK backend with agent configuration.

    Args:
        agent_config: Agent configuration with model and instructions.
    """
    self._factory = AgentFactory(agent_config=agent_config)

create_session() async

Create a new stateful multi-turn session.

Returns:

Type Description
AgentSession

An SKSession instance bound to a fresh thread run.
Source code in src/holodeck/lib/backends/sk_backend.py 
140
141
142
143
144
145
146
147
async def create_session(self) -> AgentSession:
    """Create a new stateful multi-turn session.

    Returns:
        An SKSession instance bound to a fresh thread run.
    """
    thread_run = await self._factory.create_thread_run()
    return SKSession(thread_run=thread_run)

initialize() async

Prepare the backend for use by initializing tools.

Source code in src/holodeck/lib/backends/sk_backend.py 
111
112
113
async def initialize(self) -> None:
    """Prepare the backend for use by initializing tools."""
    await self._factory._ensure_tools_initialized()

invoke_once(message, context=None) async

Execute a single stateless agent turn.

Parameters:

Name Type Description Default
message str

The user message to send to the agent.

 required
context list[dict[str, Any]] | None

Optional list of prior conversation turns (unused for now).

 None

Returns:

Type Description
ExecutionResult

ExecutionResult containing the agent response and metadata.
Source code in src/holodeck/lib/backends/sk_backend.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
async def invoke_once(
    self,
    message: str,
    context: list[dict[str, Any]] | None = None,
) -> ExecutionResult:
    """Execute a single stateless agent turn.

    Args:
        message: The user message to send to the agent.
        context: Optional list of prior conversation turns (unused for now).

    Returns:
        ExecutionResult containing the agent response and metadata.
    """
    thread_run = await self._factory.create_thread_run()
    result = await thread_run.invoke(message)
    response = _extract_response(result.chat_history)
    token_usage = result.token_usage if result.token_usage else TokenUsage.zero()
    return ExecutionResult(
        response=response,
        tool_calls=result.tool_calls,
        tool_results=result.tool_results,
        token_usage=token_usage,
    )

teardown() async

Release all backend resources.

Source code in src/holodeck/lib/backends/sk_backend.py 
149
150
151
async def teardown(self) -> None:
    """Release all backend resources."""
    await self._factory.shutdown()

SKSession

SKSession(thread_run)

Stateful multi-turn session backed by an AgentThreadRun.

Implements the AgentSession protocol by delegating to the underlying Semantic Kernel thread run for conversation management.

Initialize session with an AgentThreadRun.

Parameters:

Name Type Description Default
thread_run Any

An AgentThreadRun instance from AgentFactory.

 required
Source code in src/holodeck/lib/backends/sk_backend.py 
49
50
51
52
53
54
55
def __init__(self, thread_run: Any) -> None:
    """Initialize session with an AgentThreadRun.

    Args:
        thread_run: An AgentThreadRun instance from AgentFactory.
    """
    self._thread_run = thread_run

close() async

Release session resources. No-op for SK sessions.

Source code in src/holodeck/lib/backends/sk_backend.py 
91
92
93
async def close(self) -> None:
    """Release session resources. No-op for SK sessions."""
    pass

send(message) async

Send a message and receive a single-turn result.

Parameters:

Name Type Description Default
message str

The user message to send to the agent.

 required

Returns:

Type Description
ExecutionResult

ExecutionResult containing the agent response and metadata.
Source code in src/holodeck/lib/backends/sk_backend.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
async def send(self, message: str) -> ExecutionResult:
    """Send a message and receive a single-turn result.

    Args:
        message: The user message to send to the agent.

    Returns:
        ExecutionResult containing the agent response and metadata.
    """
    result = await self._thread_run.invoke(message)
    response = _extract_response(result.chat_history)
    token_usage = result.token_usage if result.token_usage else TokenUsage.zero()
    return ExecutionResult(
        response=response,
        tool_calls=result.tool_calls,
        tool_results=result.tool_results,
        token_usage=token_usage,
    )

send_streaming(message) async

Send a message and stream the response.

Currently delegates to send() and yields the full response as a single chunk. True streaming will be implemented in a future phase.

Parameters:

Name Type Description Default
message str

The user message to send to the agent.

 required

Yields:

Type Description
AsyncGenerator[str, None]

String chunks of the agent response.
Source code in src/holodeck/lib/backends/sk_backend.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
async def send_streaming(self, message: str) -> AsyncGenerator[str, None]:
    """Send a message and stream the response.

    Currently delegates to send() and yields the full response as a
    single chunk. True streaming will be implemented in a future phase.

    Args:
        message: The user message to send to the agent.

    Yields:
        String chunks of the agent response.
    """
    result = await self.send(message)
    yield result.response

holodeck.lib.backends.claude_backend -- Claude Agent SDK Backend

Implements the backend for provider: anthropic. Single-turn invocations use the top-level query() SDK function; multi-turn chat sessions use ClaudeSDKClient.

ClaudeBackend

ClaudeBackend(agent, tool_instances=None, mode='test', allow_side_effects=False)

Backend implementation for the Claude Agent SDK.

Implements the AgentBackend protocol. Single-turn invocations use the top-level query() function. Multi-turn sessions use ClaudeSession wrapping ClaudeSDKClient.

The constructor stores config only — no I/O, no subprocess spawned. Initialization is deferred to initialize() (called lazily on first use).

Store configuration without performing any I/O.

Parameters:

Name Type Description Default
agent Agent

Agent configuration.

 required
tool_instances dict[str, Any] | None

Initialized vectorstore/hierarchical-doc tool instances.

 None
mode str

Execution mode ("test" or "chat").

 'test'
allow_side_effects bool

Allow bash/file_system.write in test mode.

 False
Source code in src/holodeck/lib/backends/claude_backend.py 
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
def __init__(
    self,
    agent: Agent,
    tool_instances: dict[str, Any] | None = None,
    mode: str = "test",
    allow_side_effects: bool = False,
) -> None:
    """Store configuration without performing any I/O.

    Args:
        agent: Agent configuration.
        tool_instances: Initialized vectorstore/hierarchical-doc tool instances.
        mode: Execution mode (``"test"`` or ``"chat"``).
        allow_side_effects: Allow bash/file_system.write in test mode.
    """
    self._agent = agent
    self._tool_instances = tool_instances or {}
    self._mode = mode
    self._allow_side_effects = allow_side_effects
    self._initialized = False
    self._options: ClaudeAgentOptions | None = None
    self._owned_tools: list[Any] = []  # Tools created during initialize()
    self._instrumentor: Any = None

create_session() async

Create a new multi-turn session.

Automatically initializes if not yet done.

Returns:

Type Description
ClaudeSession

A new ClaudeSession instance.
Source code in src/holodeck/lib/backends/claude_backend.py 
948
949
950
951
952
953
954
955
956
957
958
959
async def create_session(self) -> ClaudeSession:
    """Create a new multi-turn session.

    Automatically initializes if not yet done.

    Returns:
        A new ``ClaudeSession`` instance.
    """
    await self._ensure_initialized()
    if self._options is None:
        raise BackendInitError("Backend options not set after initialization")
    return ClaudeSession(options=self._options)

initialize() async

Initialize the backend — validate config, build options.

Idempotent: calling multiple times is a no-op after the first.

Raises:

Type Description
BackendInitError

On validation or configuration failure.
Source code in src/holodeck/lib/backends/claude_backend.py 
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
async def initialize(self) -> None:
    """Initialize the backend — validate config, build options.

    Idempotent: calling multiple times is a no-op after the first.

    Raises:
        BackendInitError: On validation or configuration failure.
    """
    if self._initialized:
        return

    try:
        agent = self._agent
        claude = agent.claude

        # 1. Node.js prerequisite
        validate_nodejs()

        # 2. Credentials
        auth_env = validate_credentials(agent.model)

        # 3. Embedding provider (vectorstore tools)
        validate_embedding_provider(agent)

        # 4. Tool filtering (warning only)
        validate_tool_filtering(agent)

        # 4b. Auto-initialize vectorstore/hierarchical-doc tools if needed
        await self._initialize_tools()

        # 5. Tool adapters
        adapters = create_tool_adapters(
            tool_configs=agent.tools or [],
            tool_instances=self._tool_instances,
        )
        tool_server, tool_names = build_holodeck_sdk_server(adapters)

        # 6. External MCP configs
        mcp_tools = [t for t in (agent.tools or []) if isinstance(t, MCPTool)]
        mcp_configs = build_claude_mcp_configs(mcp_tools)

        # 7. OTel env vars
        otel_env: dict[str, str] = {}
        if agent.observability:
            otel_env = translate_observability(agent.observability)

        # 8. Build options
        self._options = build_options(
            agent=agent,
            tool_server=tool_server if adapters else None,
            tool_names=tool_names,
            mcp_configs=mcp_configs,
            auth_env=auth_env,
            otel_env=otel_env,
            mode=self._mode,
            allow_side_effects=self._allow_side_effects,
        )

        # 9. Working directory collision check
        wd = claude.working_directory if claude else None
        validate_working_directory(wd)

        # 10. Response format validation
        validate_response_format(agent.response_format)

        # 11. GenAI instrumentation (optional, non-blocking)
        self._activate_instrumentation()

        self._initialized = True

    except Exception as exc:
        if not isinstance(exc, BackendInitError):
            raise BackendInitError(
                f"Claude backend initialization failed: {exc}"
            ) from exc
        raise

invoke_once(message, context=None) async

Invoke the agent for a single turn.

Automatically initializes if not yet done. Retries on ProcessError (subprocess crash) with exponential backoff.

Parameters:

Name Type Description Default
message str

User message text.

 required
context list[dict[str, Any]] | None

Optional conversation context (unused for Claude backend).

 None

Returns:

Type Description
ExecutionResult

ExecutionResult with the agent's response.

Raises:

Type Description
BackendSessionError

After max retries exhausted.
Source code in src/holodeck/lib/backends/claude_backend.py 
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
async def invoke_once(
    self, message: str, context: list[dict[str, Any]] | None = None
) -> ExecutionResult:
    """Invoke the agent for a single turn.

    Automatically initializes if not yet done. Retries on ``ProcessError``
    (subprocess crash) with exponential backoff.

    Args:
        message: User message text.
        context: Optional conversation context (unused for Claude backend).

    Returns:
        ``ExecutionResult`` with the agent's response.

    Raises:
        BackendSessionError: After max retries exhausted.
    """
    await self._ensure_initialized()

    last_error: BaseException | None = None
    for attempt in range(_MAX_RETRIES):
        try:
            return await self._invoke_query(message)
        except (ProcessError, CLIConnectionError, MessageParseError) as exc:
            last_error = exc
        except BaseExceptionGroup as exc:
            # anyio TaskGroup wraps subprocess errors in ExceptionGroup
            last_error = exc

        if attempt < _MAX_RETRIES - 1:
            backoff = _BACKOFF_BASE_SECONDS * (2**attempt)
            logger.warning(
                "Claude subprocess error (attempt %d/%d), retrying in %ds: %s",
                attempt + 1,
                _MAX_RETRIES,
                backoff,
                last_error,
            )
            await asyncio.sleep(backoff)

    raise BackendSessionError(
        f"Claude subprocess failed after {_MAX_RETRIES} retries: {last_error}"
    )

teardown() async

Reset backend state, releasing any built options.

Source code in src/holodeck/lib/backends/claude_backend.py 
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
async def teardown(self) -> None:
    """Reset backend state, releasing any built options."""
    # Deactivate GenAI instrumentation
    if self._instrumentor is not None:
        try:
            self._instrumentor.uninstrument()
        except Exception as exc:
            logger.warning("Error deactivating GenAI instrumentation: %s", exc)
        self._instrumentor = None

    # Cleanup owned tools
    for tool_inst in self._owned_tools:
        if hasattr(tool_inst, "cleanup"):
            try:
                await tool_inst.cleanup()
            except Exception as exc:
                logger.warning("Error cleaning up tool: %s", exc)
    self._owned_tools = []
    self._tool_instances = {}
    self._initialized = False
    self._options = None

ClaudeSession

ClaudeSession(options)

Stateful multi-turn session backed by ClaudeSDKClient.

Each session maintains a session_id across turns. Multi-turn state is opt-in: after the first turn, subsequent turns pass continue_conversation=True and resume=session_id to the SDK.

The _base_options reference is never mutated. Turn-specific options are created as new ClaudeAgentOptions instances.

Initialize session with base options.

Parameters:

Name Type Description Default
options ClaudeAgentOptions

Base options (immutable reference for the session lifetime).

 required
Source code in src/holodeck/lib/backends/claude_backend.py 
488
489
490
491
492
493
494
495
496
497
498
def __init__(self, options: ClaudeAgentOptions) -> None:
    """Initialize session with base options.

    Args:
        options: Base options (immutable reference for the session lifetime).
    """
    self._base_options = options
    self._session_id: str | None = None
    self._client: ClaudeSDKClient | None = None
    self._turn_count: int = 0
    self._tool_event_queue: asyncio.Queue[ToolEvent] = asyncio.Queue()

tool_events property

Queue of real-time tool events emitted via SDK hooks.

close() async

Disconnect the SDK client and release resources.

Source code in src/holodeck/lib/backends/claude_backend.py 
658
659
660
async def close(self) -> None:
    """Disconnect the SDK client and release resources."""
    await self.release_transport()

release_transport() async

Disconnect the SDK client without losing session state.

After calling this, the next send() or send_streaming() call will create a fresh ClaudeSDKClient and reconnect, resuming the conversation via the preserved session_id.

This is required when the session is used across different async task contexts (e.g., HTTP requests in holodeck serve), because the SDK's anyio task group is bound to the task that called connect().

Source code in src/holodeck/lib/backends/claude_backend.py 
643
644
645
646
647
648
649
650
651
652
653
654
655
656
async def release_transport(self) -> None:
    """Disconnect the SDK client without losing session state.

    After calling this, the next ``send()`` or ``send_streaming()`` call
    will create a fresh ``ClaudeSDKClient`` and reconnect, resuming the
    conversation via the preserved ``session_id``.

    This is required when the session is used across different async task
    contexts (e.g., HTTP requests in ``holodeck serve``), because the
    SDK's anyio task group is bound to the task that called ``connect()``.
    """
    if self._client is not None:
        await self._client.disconnect()
        self._client = None

send(message) async

Send a message and collect the full response.

Parameters:

Name Type Description Default
message str

User message text.

 required

Returns:

Type Description
ExecutionResult

ExecutionResult with the agent's response.

Raises:

Type Description
BackendSessionError

On subprocess or SDK error.
Source code in src/holodeck/lib/backends/claude_backend.py 
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
async def send(self, message: str) -> ExecutionResult:
    """Send a message and collect the full response.

    Args:
        message: User message text.

    Returns:
        ``ExecutionResult`` with the agent's response.

    Raises:
        BackendSessionError: On subprocess or SDK error.
    """
    try:
        client = await self._ensure_client()
        await client.query(message, session_id=self._get_session_id())

        text_parts: list[str] = []
        tool_calls: list[dict[str, Any]] = []
        tool_results: list[dict[str, Any]] = []
        token_usage = TokenUsage.zero()
        num_turns = 1

        async for msg in client.receive_response():
            text_parts, tool_calls, tool_results = _process_message(
                msg, text_parts, tool_calls, tool_results
            )
            if msg.__class__.__name__ == "ResultMessage":
                rm = cast(Any, msg)
                usage = rm.usage or {}
                prompt = usage.get("input_tokens", 0)
                completion = usage.get("output_tokens", 0)
                token_usage = TokenUsage(
                    prompt_tokens=prompt,
                    completion_tokens=completion,
                    total_tokens=prompt + completion,
                )
                num_turns = rm.num_turns
                self._session_id = rm.session_id

        self._turn_count += 1

        return ExecutionResult(
            response="".join(text_parts),
            tool_calls=tool_calls,
            tool_results=tool_results,
            token_usage=token_usage,
            num_turns=num_turns,
        )
    except (ProcessError, CLIConnectionError) as exc:
        raise BackendSessionError(
            f"subprocess terminated unexpectedly: {exc}"
        ) from exc

send_streaming(message) async

Send a message and yield text chunks progressively.

Parameters:

Name Type Description Default
message str

User message text.

 required

Yields:

Type Description
AsyncGenerator[str, None]

Text chunks as they arrive from the SDK.

Raises:

Type Description
BackendSessionError

On subprocess or SDK error.
Source code in src/holodeck/lib/backends/claude_backend.py 
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
async def send_streaming(self, message: str) -> AsyncGenerator[str, None]:
    """Send a message and yield text chunks progressively.

    Args:
        message: User message text.

    Yields:
        Text chunks as they arrive from the SDK.

    Raises:
        BackendSessionError: On subprocess or SDK error.
    """
    try:
        client = await self._ensure_client()
        await client.query(message, session_id=self._get_session_id())

        async for msg in client.receive_response():
            if msg.__class__.__name__ == "AssistantMessage":
                for block in cast(Any, msg).content:
                    if block.__class__.__name__ == "TextBlock" and block.text:
                        yield block.text
            elif msg.__class__.__name__ == "ResultMessage":
                rm = cast(Any, msg)
                self._session_id = rm.session_id
                self._turn_count += 1
    except (ProcessError, CLIConnectionError) as exc:
        raise BackendSessionError(
            f"subprocess terminated unexpectedly: {exc}"
        ) from exc

build_options

build_options(*, agent, tool_server, tool_names, mcp_configs, auth_env, otel_env, mode, allow_side_effects)

Assemble ClaudeAgentOptions from agent config and bridge outputs.

Parameters:

Name Type Description Default
agent Agent

The agent configuration.

 required
tool_server McpSdkServerConfig | None

In-process MCP server for vectorstore/hierarchical-doc tools.

 required
tool_names list[str]

Allowed tool names from the in-process server.

 required
mcp_configs dict[str, Any]

External MCP server configs from build_claude_mcp_configs().

 required
auth_env dict[str, str]

Auth env vars from validate_credentials().

 required
otel_env dict[str, str]

OTel env vars from translate_observability().

 required
mode str

Execution mode ("test" or "chat").

 required
allow_side_effects bool

Whether side effects are allowed in test mode.

 required

Returns:

Type Description
ClaudeAgentOptions

Configured ClaudeAgentOptions ready for query() or ClaudeSDKClient.
Source code in src/holodeck/lib/backends/claude_backend.py 
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
def build_options(
    *,
    agent: Agent,
    tool_server: McpSdkServerConfig | None,
    tool_names: list[str],
    mcp_configs: dict[str, Any],
    auth_env: dict[str, str],
    otel_env: dict[str, str],
    mode: str,
    allow_side_effects: bool,
) -> ClaudeAgentOptions:
    """Assemble ``ClaudeAgentOptions`` from agent config and bridge outputs.

    Args:
        agent: The agent configuration.
        tool_server: In-process MCP server for vectorstore/hierarchical-doc tools.
        tool_names: Allowed tool names from the in-process server.
        mcp_configs: External MCP server configs from ``build_claude_mcp_configs()``.
        auth_env: Auth env vars from ``validate_credentials()``.
        otel_env: OTel env vars from ``translate_observability()``.
        mode: Execution mode (``"test"`` or ``"chat"``).
        allow_side_effects: Whether side effects are allowed in test mode.

    Returns:
        Configured ``ClaudeAgentOptions`` ready for ``query()`` or ``ClaudeSDKClient``.
    """
    claude = agent.claude
    system_prompt = resolve_instructions(agent.instructions)

    # MCP servers
    mcp_servers: dict[str, Any] = dict(mcp_configs)
    if tool_server is not None:
        mcp_servers["holodeck_tools"] = tool_server

    # Env vars — unset CLAUDECODE to prevent the "nested session" guard
    # when HoloDeck runs inside a terminal with Claude Code active.
    # SDK merges: {**os.environ, **options.env}, so "" overrides "1".
    env: dict[str, str] = {"CLAUDECODE": "", **auth_env, **otel_env}

    # Permission mode
    perm_mode = None
    if claude is not None:
        perm_mode = _build_permission_mode(
            claude.permission_mode, mode, allow_side_effects
        )

    # Extended thinking
    max_thinking_tokens = None
    if claude and claude.extended_thinking and claude.extended_thinking.enabled:
        max_thinking_tokens = claude.extended_thinking.budget_tokens

    # Allowed tools
    allowed_tools: list[str] = list(tool_names)
    if claude and claude.allowed_tools:
        allowed_tools.extend(claude.allowed_tools)

    # Built-in capabilities
    if claude and claude.web_search:
        allowed_tools.append("WebSearch")

    # Output format
    output_format = _build_output_format(agent.response_format)

    # Working directory — fall back to agent.yaml's directory so that
    # relative paths in MCP args (e.g. "./data") resolve correctly.
    from holodeck.config.context import agent_base_dir

    cwd = claude.working_directory if claude else None
    if cwd is None:
        cwd = agent_base_dir.get()

    # Max turns
    max_turns = claude.max_turns if claude else None

    # Build the options dict
    opts_kwargs: dict[str, Any] = {
        "model": agent.model.name,
        "system_prompt": system_prompt,
        "permission_mode": perm_mode,
        "max_turns": max_turns,
        "mcp_servers": mcp_servers,
        "allowed_tools": allowed_tools,
        "env": env,
        "cwd": cwd,
        "output_format": output_format,
    }

    if max_thinking_tokens is not None:
        opts_kwargs["max_thinking_tokens"] = max_thinking_tokens

    return ClaudeAgentOptions(**opts_kwargs)

holodeck.lib.backends.tool_adapters -- Claude SDK Tool Adapters

Wraps HoloDeck vectorstore and hierarchical-document tools as @tool-decorated functions, bundles them into an in-process MCP server, and provides a factory for ClaudeBackend to call during initialization.

VectorStoreToolAdapter

VectorStoreToolAdapter(config, instance)

Wraps a VectorStoreTool for use with the Claude Agent SDK.

Parameters:

Name Type Description Default
config VectorstoreTool

The vectorstore tool configuration from the agent YAML.

 required
instance VectorStoreTool

An initialized VectorStoreTool instance.

 required
Source code in src/holodeck/lib/backends/tool_adapters.py 
 98
 99
100
101
102
103
104
def __init__(
    self,
    config: VectorstoreToolConfig,
    instance: VectorStoreTool,
) -> None:
    self.config = config
    self.instance = instance

to_sdk_tool()

Return an SdkMcpTool backed by this adapter's search method.

Source code in src/holodeck/lib/backends/tool_adapters.py 
106
107
108
109
110
111
112
def to_sdk_tool(self) -> SdkMcpTool[Any]:
    """Return an ``SdkMcpTool`` backed by this adapter's search method."""
    name = f"{self.config.name}_search"
    desc = _truncate_description(
        f"Search {self.config.name}: {self.config.description}"
    )
    return _make_vectorstore_search_fn(self.instance, name, desc)

HierarchicalDocToolAdapter

HierarchicalDocToolAdapter(config, instance)

Wraps a HierarchicalDocumentTool for use with the Claude Agent SDK.

Parameters:

Name Type Description Default
config HierarchicalDocumentToolConfig

The hierarchical document tool configuration from the agent YAML.

 required
instance HierarchicalDocumentTool

An initialized HierarchicalDocumentTool instance.

 required
Source code in src/holodeck/lib/backends/tool_adapters.py 
123
124
125
126
127
128
129
def __init__(
    self,
    config: HierarchicalDocumentToolConfig,
    instance: HierarchicalDocumentTool,
) -> None:
    self.config = config
    self.instance = instance

to_sdk_tool()

Return an SdkMcpTool backed by this adapter's search method.

Source code in src/holodeck/lib/backends/tool_adapters.py 
131
132
133
134
135
136
137
def to_sdk_tool(self) -> SdkMcpTool[Any]:
    """Return an ``SdkMcpTool`` backed by this adapter's search method."""
    name = f"{self.config.name}_search"
    desc = _truncate_description(
        f"Search {self.config.name}: {self.config.description}"
    )
    return _make_hierarchical_search_fn(self.instance, name, desc)

create_tool_adapters

create_tool_adapters(tool_configs, tool_instances)

Build adapters for vectorstore and hierarchical-document tools.

Filters tool_configs for supported types, matches each to its initialized instance by config.name, and returns adapter objects.

Parameters:

Name Type Description Default
tool_configs list[ToolUnion]

All tool configurations from the agent YAML.

 required
tool_instances dict[str, VectorStoreTool | HierarchicalDocumentTool]

Initialized tool instances keyed by config name.

 required

Returns:

Type Description
list[VectorStoreToolAdapter | HierarchicalDocToolAdapter]

List of adapter objects ready for build_holodeck_sdk_server().

Raises:

Type Description
BackendInitError

If a supported tool config has no matching instance.
Source code in src/holodeck/lib/backends/tool_adapters.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
def create_tool_adapters(
    tool_configs: list[ToolUnion],
    tool_instances: dict[str, VectorStoreTool | HierarchicalDocumentTool],
) -> list[VectorStoreToolAdapter | HierarchicalDocToolAdapter]:
    """Build adapters for vectorstore and hierarchical-document tools.

    Filters *tool_configs* for supported types, matches each to its
    initialized instance by ``config.name``, and returns adapter objects.

    Args:
        tool_configs: All tool configurations from the agent YAML.
        tool_instances: Initialized tool instances keyed by config name.

    Returns:
        List of adapter objects ready for ``build_holodeck_sdk_server()``.

    Raises:
        BackendInitError: If a supported tool config has no matching instance.
    """
    adapters: list[VectorStoreToolAdapter | HierarchicalDocToolAdapter] = []

    for cfg in tool_configs:
        if isinstance(cfg, VectorstoreToolConfig):
            instance = tool_instances.get(cfg.name)
            if instance is None:
                raise BackendInitError(
                    f"No initialized instance found for tool '{cfg.name}' "
                    f"(type: {cfg.type}). Ensure tool initialization "
                    "completed before creating adapters."
                )
            adapters.append(
                VectorStoreToolAdapter(
                    config=cfg,
                    instance=instance,  # type: ignore[arg-type]
                )
            )
        elif isinstance(cfg, HierarchicalDocumentToolConfig):
            instance = tool_instances.get(cfg.name)
            if instance is None:
                raise BackendInitError(
                    f"No initialized instance found for tool '{cfg.name}' "
                    f"(type: {cfg.type}). Ensure tool initialization "
                    "completed before creating adapters."
                )
            adapters.append(
                HierarchicalDocToolAdapter(
                    config=cfg,
                    instance=instance,  # type: ignore[arg-type]
                )
            )

    return adapters

build_holodeck_sdk_server

build_holodeck_sdk_server(adapters)

Bundle adapters into an in-process MCP server for the Claude subprocess.

Parameters:

Name Type Description Default
adapters list[VectorStoreToolAdapter | HierarchicalDocToolAdapter]

Adapter objects produced by create_tool_adapters().

 required

Returns:

Type Description
McpSdkServerConfig

A tuple of (server_config, allowed_tool_names) where
list[str]

server_config is a McpSdkServerConfig TypedDict and
tuple[McpSdkServerConfig, list[str]]

allowed_tool_names are the fully-qualified MCP tool names.
Source code in src/holodeck/lib/backends/tool_adapters.py 
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
def build_holodeck_sdk_server(
    adapters: list[VectorStoreToolAdapter | HierarchicalDocToolAdapter],
) -> tuple[McpSdkServerConfig, list[str]]:
    """Bundle adapters into an in-process MCP server for the Claude subprocess.

    Args:
        adapters: Adapter objects produced by ``create_tool_adapters()``.

    Returns:
        A tuple of ``(server_config, allowed_tool_names)`` where
        *server_config* is a ``McpSdkServerConfig`` TypedDict and
        *allowed_tool_names* are the fully-qualified MCP tool names.
    """
    sdk_tools: list[SdkMcpTool[Any]] = [a.to_sdk_tool() for a in adapters]

    server_config: McpSdkServerConfig = create_sdk_mcp_server(
        name=_SERVER_NAME,
        tools=sdk_tools,
    )

    allowed_tools = [f"mcp__{_SERVER_NAME}__{a.config.name}_search" for a in adapters]

    return server_config, allowed_tools

holodeck.lib.backends.mcp_bridge -- MCP Configuration Bridge

Translates HoloDeck MCPTool configurations into Claude Agent SDK McpStdioServerConfig format for subprocess-based MCP servers. Only stdio transport tools are supported.

build_claude_mcp_configs

build_claude_mcp_configs(mcp_tools)

Translate HoloDeck MCPTool configs to Claude SDK MCP server configs.

Only stdio transport tools are supported by the Claude subprocess. Non-stdio tools are skipped with a warning.

Parameters:

Name Type Description Default
mcp_tools list[MCPTool]

List of MCPTool configurations from agent YAML.

 required

Returns:

Type Description
dict[str, McpStdioServerConfig]

Dictionary mapping tool names to McpStdioServerConfig TypedDicts.
Source code in src/holodeck/lib/backends/mcp_bridge.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def build_claude_mcp_configs(
    mcp_tools: list[MCPTool],
) -> dict[str, McpStdioServerConfig]:
    """Translate HoloDeck MCPTool configs to Claude SDK MCP server configs.

    Only stdio transport tools are supported by the Claude subprocess.
    Non-stdio tools are skipped with a warning.

    Args:
        mcp_tools: List of MCPTool configurations from agent YAML.

    Returns:
        Dictionary mapping tool names to McpStdioServerConfig TypedDicts.
    """
    from holodeck.config.context import agent_base_dir

    base_dir = agent_base_dir.get()

    configs: dict[str, McpStdioServerConfig] = {}

    for tool in mcp_tools:
        if tool.transport != TransportType.STDIO:
            logger.warning(
                "Skipping MCP tool '%s': %s transport is not supported by "
                "Claude subprocess (only stdio is supported)",
                tool.name,
                tool.transport.value,
            )
            continue

        command = tool.command.value if tool.command else "npx"
        args = _resolve_relative_args(tool.args or [], base_dir)
        env = _resolve_mcp_env(tool)

        entry: McpStdioServerConfig = {
            "command": command,
            "args": args,
        }

        if env:
            entry["env"] = env

        configs[tool.name] = entry

    return configs

holodeck.lib.backends.otel_bridge -- Observability Bridge

Translates HoloDeck ObservabilityConfig into environment variable dicts that configure OpenTelemetry for the Claude subprocess.

translate_observability

translate_observability(config)

Translate ObservabilityConfig to env vars for the Claude subprocess.

Produces a dict of environment variable key-value pairs that configure OpenTelemetry in the Claude subprocess. All values are strings.

Parameters:

Name Type Description Default
config ObservabilityConfig

HoloDeck observability configuration from agent YAML.

 required

Returns:

Type Description
dict[str, str]

Dictionary of environment variable names to string values.
dict[str, str]

Empty dict if observability is disabled.
Source code in src/holodeck/lib/backends/otel_bridge.py 
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def translate_observability(config: ObservabilityConfig) -> dict[str, str]:
    """Translate ObservabilityConfig to env vars for the Claude subprocess.

    Produces a dict of environment variable key-value pairs that configure
    OpenTelemetry in the Claude subprocess. All values are strings.

    Args:
        config: HoloDeck observability configuration from agent YAML.

    Returns:
        Dictionary of environment variable names to string values.
        Empty dict if observability is disabled.
    """
    if not config.enabled:
        return {}

    env: dict[str, str] = {}

    # Core telemetry enablement
    env["CLAUDE_CODE_ENABLE_TELEMETRY"] = "1"

    # Disable subprocess trace export when Python-side GenAI instrumentation
    # is active.  The instrumentor's hooks produce the authoritative
    # invoke_agent / execute_tool spans in-process; letting the subprocess
    # also export traces creates duplicate, unlinked trace trees in the
    # collector and confuses dashboards like Aspire.
    if config.traces.enabled:
        env["OTEL_TRACES_EXPORTER"] = "none"

    # OTLP exporter configuration
    otlp = config.exporters.otlp
    otlp_enabled = otlp is not None and otlp.enabled

    # Metrics exporter
    if otlp_enabled and config.metrics.enabled:
        env["OTEL_METRICS_EXPORTER"] = "otlp"
    else:
        env["OTEL_METRICS_EXPORTER"] = "none"

    # Logs exporter
    if otlp_enabled and config.logs.enabled:
        env["OTEL_LOGS_EXPORTER"] = "otlp"
    else:
        env["OTEL_LOGS_EXPORTER"] = "none"

    # Protocol and endpoint (only if OTLP is enabled)
    if otlp_enabled and otlp is not None:
        env["OTEL_EXPORTER_OTLP_PROTOCOL"] = _PROTOCOL_MAP[otlp.protocol]
        env["OTEL_EXPORTER_OTLP_ENDPOINT"] = otlp.endpoint

    # Export intervals
    interval = str(config.metrics.export_interval_ms)
    env["OTEL_METRIC_EXPORT_INTERVAL"] = interval
    env["OTEL_LOGS_EXPORT_INTERVAL"] = interval

    # Privacy controls (FR-038: default off)
    if config.traces.capture_content:
        env["OTEL_LOG_USER_PROMPTS"] = "true"
        env["OTEL_LOG_TOOL_DETAILS"] = "true"

    # Warn about unsupported fields
    unsupported = _collect_unsupported_fields(config)
    if unsupported:
        logger.warning(
            "Claude subprocess does not support these observability settings "
            "(they will be ignored): %s",
            ", ".join(unsupported),
        )

    return env

holodeck.lib.backends.validators -- Startup Validators

Pre-flight checks called by ClaudeBackend.initialize() before spawning the Claude subprocess. These surface configuration errors at startup rather than at runtime.

validate_nodejs

validate_nodejs()

Validate that Node.js is available on PATH.

Claude Agent SDK requires Node.js to spawn its subprocess.

Raises:

Type Description
ConfigError

If node is not found on PATH.
Source code in src/holodeck/lib/backends/validators.py 
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def validate_nodejs() -> None:
    """Validate that Node.js is available on PATH.

    Claude Agent SDK requires Node.js to spawn its subprocess.

    Raises:
        ConfigError: If node is not found on PATH.
    """
    if shutil.which("node") is None:
        raise ConfigError(
            "nodejs",
            "Node.js is required to run Claude Agent SDK but was not found on "
            "PATH. Install Node.js from https://nodejs.org/ and ensure it is "
            "on your PATH.",
        )

    stdout = ""
    try:
        result = subprocess.run(  # noqa: S603 # nosec B603 B607
            ["node", "--version"],  # noqa: S607
            capture_output=True,
            text=True,
            timeout=5,
        )
        stdout = result.stdout.strip()
        version = stdout.lstrip("v")
        major = int(version.split(".")[0])
    except subprocess.TimeoutExpired as e:
        raise ConfigError(
            "nodejs",
            "Node.js version check timed out. Ensure 'node --version' "
            "completes quickly.",
        ) from e
    except subprocess.CalledProcessError as e:
        raise ConfigError(
            "nodejs",
            "Failed to determine Node.js version. Ensure 'node --version' " "works.",
        ) from e
    except ValueError as e:
        raise ConfigError(
            "nodejs",
            f"Could not parse Node.js version from output: {stdout}",
        ) from e

    if major < NODEJS_MIN_VERSION:
        raise ConfigError(
            "nodejs",
            f"Node.js version {version} found but >= {NODEJS_MIN_VERSION} is "
            "required by Claude Agent SDK. Download from https://nodejs.org/",
        )

    logger.debug(f"Node.js version {version} validated (>= {NODEJS_MIN_VERSION})")

validate_credentials

validate_credentials(model)

Validate authentication credentials for the LLM provider.

Checks that the required environment variables are present for the configured auth_provider, including cloud routing context for Bedrock, Vertex, and Foundry. Returns a dict of environment variables to inject into the Claude subprocess.

Parameters:

Name Type Description Default
model LLMProvider

LLM provider configuration.

 required

Returns:

Type Description
dict[str, str]

Dict of environment variables to set for the subprocess.

Raises:

Type Description
ConfigError

If required credentials are absent.
Source code in src/holodeck/lib/backends/validators.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
def validate_credentials(model: LLMProvider) -> dict[str, str]:
    """Validate authentication credentials for the LLM provider.

    Checks that the required environment variables are present for the
    configured auth_provider, including cloud routing context for Bedrock,
    Vertex, and Foundry. Returns a dict of environment variables to inject
    into the Claude subprocess.

    Args:
        model: LLM provider configuration.

    Returns:
        Dict of environment variables to set for the subprocess.

    Raises:
        ConfigError: If required credentials are absent.
    """
    auth = model.auth_provider or AuthProvider.api_key

    if auth == AuthProvider.api_key:
        key = _get_required_env_var(
            "ANTHROPIC_API_KEY",
            "ANTHROPIC_API_KEY environment variable is not set. "
            "Set it with: export ANTHROPIC_API_KEY=sk-ant-...",
        )
        return {"ANTHROPIC_API_KEY": key}

    if auth == AuthProvider.oauth_token:
        token = _get_required_env_var(
            "CLAUDE_CODE_OAUTH_TOKEN",
            "CLAUDE_CODE_OAUTH_TOKEN environment variable is not set. "
            "Run `claude setup-token` to authenticate with OAuth.",
        )
        return {"CLAUDE_CODE_OAUTH_TOKEN": token}

    if auth == AuthProvider.bedrock:
        bedrock_region = _get_first_present_env_var(_BEDROCK_REGION_ENV_CANDIDATES)
        if bedrock_region is None:
            raise ConfigError(
                "AWS_REGION",
                "Missing AWS region for auth_provider: bedrock. Set either "
                "AWS_REGION or AWS_DEFAULT_REGION (for example: "
                "export AWS_REGION=us-east-1).",
            )
        return {
            "CLAUDE_CODE_USE_BEDROCK": "1",
            bedrock_region[0]: bedrock_region[1],
        }

    if auth == AuthProvider.vertex:
        region = _get_required_env_var(
            "CLOUD_ML_REGION",
            "CLOUD_ML_REGION environment variable is not set for "
            "auth_provider: vertex. Set it with: "
            "export CLOUD_ML_REGION=us-east5",
        )
        project_context = _get_first_present_env_var(_VERTEX_PROJECT_ENV_CANDIDATES)
        if project_context is None:
            raise ConfigError(
                "ANTHROPIC_VERTEX_PROJECT_ID",
                "Missing Vertex project context for auth_provider: vertex. "
                "Set one of: ANTHROPIC_VERTEX_PROJECT_ID, GCLOUD_PROJECT, "
                "GOOGLE_CLOUD_PROJECT, or GOOGLE_APPLICATION_CREDENTIALS.",
            )
        return {
            "CLAUDE_CODE_USE_VERTEX": "1",
            "CLOUD_ML_REGION": region,
            project_context[0]: project_context[1],
        }

    # AuthProvider.foundry
    foundry_target = _get_first_present_env_var(_FOUNDRY_TARGET_ENV_CANDIDATES)
    if foundry_target is None:
        raise ConfigError(
            "ANTHROPIC_FOUNDRY_RESOURCE",
            "Missing Foundry target for auth_provider: foundry. "
            "Set one of: ANTHROPIC_FOUNDRY_RESOURCE or "
            "ANTHROPIC_FOUNDRY_BASE_URL.",
        )
    return {"CLAUDE_CODE_USE_FOUNDRY": "1", foundry_target[0]: foundry_target[1]}

validate_embedding_provider

validate_embedding_provider(agent)

Validate embedding provider configuration for vectorstore tools.

Anthropic does not support generating embeddings, so an external embedding_provider must be specified when using vectorstore tools with the Anthropic LLM provider.

Parameters:

Name Type Description Default
agent Agent

Agent configuration to validate.

 required

Raises:

Type Description
ConfigError

If embedding configuration is invalid for the provider.
Source code in src/holodeck/lib/backends/validators.py 
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
def validate_embedding_provider(agent: Agent) -> None:
    """Validate embedding provider configuration for vectorstore tools.

    Anthropic does not support generating embeddings, so an external
    embedding_provider must be specified when using vectorstore tools
    with the Anthropic LLM provider.

    Args:
        agent: Agent configuration to validate.

    Raises:
        ConfigError: If embedding configuration is invalid for the provider.
    """
    if not agent.tools:
        return

    has_vectorstore = any(
        isinstance(tool, VectorstoreTool | HierarchicalDocumentToolConfig)
        for tool in agent.tools
    )
    if not has_vectorstore:
        return

    # Anthropic cannot be used as an embedding provider
    if agent.embedding_provider is not None:
        if agent.embedding_provider.provider == ProviderEnum.ANTHROPIC:
            raise ConfigError(
                "embedding_provider",
                "Anthropic cannot generate embeddings. Use a different provider "
                "such as openai or azure_openai for embedding_provider when "
                "using vectorstore tools.",
            )
        return

    # Anthropic LLM + vectorstore tool + no embedding_provider
    if agent.model.provider == ProviderEnum.ANTHROPIC:
        raise ConfigError(
            "embedding_provider",
            "embedding_provider is required when using vectorstore tools with "
            "provider: anthropic. Add an embedding_provider using openai or "
            "azure_openai.",
        )

validate_tool_filtering

validate_tool_filtering(agent)

Warn if tool_filtering is configured for Anthropic provider.

Claude Agent SDK manages tool selection natively; tool_filtering is a Semantic Kernel feature that is not supported by the Claude backend.

This validator never raises — it only emits a warning. The tool_filtering field is not mutated.

Parameters:

Name Type Description Default
agent Agent

Agent configuration to validate.

 required
Source code in src/holodeck/lib/backends/validators.py 
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
def validate_tool_filtering(agent: Agent) -> None:
    """Warn if tool_filtering is configured for Anthropic provider.

    Claude Agent SDK manages tool selection natively; tool_filtering is a
    Semantic Kernel feature that is not supported by the Claude backend.

    This validator never raises — it only emits a warning. The
    tool_filtering field is not mutated.

    Args:
        agent: Agent configuration to validate.
    """
    if agent.tool_filtering is None:
        return

    if agent.model.provider == ProviderEnum.ANTHROPIC:
        logger.warning(
            "tool_filtering is configured but will be ignored when using "
            "provider: anthropic with the Claude Agent SDK backend. "
            "Claude manages tool selection natively.",
        )

validate_working_directory

validate_working_directory(path)

Warn if CLAUDE.md in working directory may conflict with agent instructions.

Detects a CLAUDE.md file that contains a '# CLAUDE.md' header, which is the standard format used by Claude Code project instructions. Such a file may override or conflict with the agent's configured instructions.

Parameters:

Name Type Description Default
path str | None

Working directory path, or None to skip validation.

 required
Source code in src/holodeck/lib/backends/validators.py 
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
def validate_working_directory(path: str | None) -> None:
    """Warn if CLAUDE.md in working directory may conflict with agent instructions.

    Detects a CLAUDE.md file that contains a '# CLAUDE.md' header, which
    is the standard format used by Claude Code project instructions. Such
    a file may override or conflict with the agent's configured instructions.

    Args:
        path: Working directory path, or None to skip validation.
    """
    if path is None:
        return

    claude_md = Path(path) / "CLAUDE.md"
    if not claude_md.exists():
        return

    content = claude_md.read_text()
    if "# CLAUDE.md" in content:
        logger.warning(
            "A CLAUDE.md file with a '# CLAUDE.md' header was found in the "
            "working directory '%s'. This may conflict with the agent's system "
            "instructions. Review the file to avoid unexpected behavior.",
            path,
        )

validate_response_format

validate_response_format(response_format)

Validate response format schema is serializable and accessible.

Parameters:

Name Type Description Default
response_format dict[str, Any] | str | None

Inline schema dict, file path string, or None.

 required

Raises:

Type Description
ConfigError

If the schema is not JSON-serializable or file not found.
Source code in src/holodeck/lib/backends/validators.py 
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
def validate_response_format(response_format: dict[str, Any] | str | None) -> None:
    """Validate response format schema is serializable and accessible.

    Args:
        response_format: Inline schema dict, file path string, or None.

    Raises:
        ConfigError: If the schema is not JSON-serializable or file not found.
    """
    if response_format is None:
        return

    if isinstance(response_format, str):
        schema_path = Path(response_format)
        if not schema_path.exists():
            raise ConfigError(
                "response_format",
                f"response_format file not found: {response_format}",
            )
        try:
            json.loads(schema_path.read_text())
        except (json.JSONDecodeError, OSError) as e:
            raise ConfigError(
                "response_format",
                f"response_format file is not valid JSON: {e}",
            ) from e
        return

    # Must be a dict — verify it is JSON-serializable
    try:
        json.dumps(response_format)
    except (TypeError, ValueError) as e:
        raise ConfigError(
            "response_format",
            f"response_format contains non-JSON-serializable values: {e}",
        ) from e