Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions src/claude_agent_sdk/_internal/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -179,6 +179,7 @@ async def _process_query_inner(
agents=agents_dict,
exclude_dynamic_sections=exclude_dynamic_sections,
skills=configured_options.skills,
tool_aliases=configured_options.tool_aliases,
)

if configured_options.session_store is not None:
Expand Down
6 changes: 6 additions & 0 deletions src/claude_agent_sdk/_internal/query.py
Original file line number Diff line number Diff line change
Expand Up @@ -84,6 +84,7 @@ def __init__(
agents: dict[str, dict[str, Any]] | None = None,
exclude_dynamic_sections: bool | None = None,
skills: list[str] | Literal["all"] | None = None,
tool_aliases: dict[str, str] | None = None,
):
"""Initialize Query with transport and callbacks.

Expand All @@ -99,6 +100,8 @@ def __init__(
initialize (see ``SystemPromptPreset``)
skills: Optional skill allowlist to send via initialize so the CLI
can filter which skills are loaded into the system prompt
tool_aliases: Optional mapping of built-in tool names to MCP tool
names sent via initialize (see ``ClaudeAgentOptions.tool_aliases``)
"""
self._initialize_timeout = initialize_timeout
self.transport = transport
Expand All @@ -109,6 +112,7 @@ def __init__(
self._agents = agents
self._exclude_dynamic_sections = exclude_dynamic_sections
self._skills = skills
self._tool_aliases = tool_aliases

# Control protocol state
self.pending_control_responses: dict[str, anyio.Event] = {}
Expand Down Expand Up @@ -212,6 +216,8 @@ async def initialize(self) -> dict[str, Any] | None:
# only send the field when it's an explicit list.
if isinstance(self._skills, list):
request["skills"] = self._skills
if self._tool_aliases:
request["toolAliases"] = self._tool_aliases

# Use longer timeout for initialize since MCP servers may take time to start
response = await self._send_control_request(
Expand Down
1 change: 1 addition & 0 deletions src/claude_agent_sdk/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -234,6 +234,7 @@ async def _connect_inner(
agents=agents_dict,
exclude_dynamic_sections=exclude_dynamic_sections,
skills=self.options.skills,
tool_aliases=self.options.tool_aliases,
)

if self.options.session_store is not None:
Expand Down
12 changes: 12 additions & 0 deletions src/claude_agent_sdk/types.py
Original file line number Diff line number Diff line change
Expand Up @@ -1894,6 +1894,17 @@ class ClaudeAgentOptions:
``{"type": "json_schema", "schema": {"type": "object", "properties": {...}}}``.
"""

tool_aliases: dict[str, str] | None = None
"""Map built-in tool names to MCP tool names.

When set, Claude calls the specified MCP tool in place of the named
built-in. For example, ``{"Bash": "mcp__workspace__bash"}`` routes every
``Bash`` call through the ``mcp__workspace__bash`` MCP tool instead.

Sent via the initialization message at session start (not a CLI flag).
Matches the TypeScript SDK's ``toolAliases`` option.
"""

enable_file_checkpointing: bool = False
"""Enable file checkpointing to track file changes during the session.

Expand Down Expand Up @@ -1962,6 +1973,7 @@ class SDKControlInitializeRequest(TypedDict):
subtype: Literal["initialize"]
hooks: dict[HookEvent, Any] | None
agents: NotRequired[dict[str, dict[str, Any]]]
toolAliases: NotRequired[dict[str, str]]


class SDKControlSetPermissionModeRequest(TypedDict):
Expand Down
128 changes: 128 additions & 0 deletions tests/test_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -364,3 +364,131 @@ async def read_messages():
receive_stream.receive_nowait()

anyio.run(_test)


class TestToolAliasesPassthrough:
"""Verify that tool_aliases in ClaudeAgentOptions reaches the Query constructor
in both the query() (InternalClient) and ClaudeSDKClient code paths."""

def _make_mock_query(self):
"""Minimal Query mock sufficient for connect/disconnect."""
mock_query = AsyncMock()
mock_query.start = AsyncMock()
mock_query.initialize = AsyncMock(return_value=None)
mock_query.close = AsyncMock()
mock_query.close_receive_stream = Mock()
mock_query._tg = None
mock_query.set_transcript_mirror_batcher = Mock()

def _consume_coro(coro):
coro.close()
return Mock()

mock_query.spawn_task = Mock(side_effect=_consume_coro)

async def mock_receive():
yield {
"type": "result",
"subtype": "success",
"duration_ms": 100,
"duration_api_ms": 80,
"is_error": False,
"num_turns": 1,
"session_id": "test",
}

mock_query.receive_messages = mock_receive
return mock_query

def _make_mock_transport(self):
mock_transport = AsyncMock()
mock_transport.connect = AsyncMock()
mock_transport.close = AsyncMock()
mock_transport.end_input = AsyncMock()
mock_transport.write = AsyncMock()
mock_transport.is_ready = Mock(return_value=True)
return mock_transport

def test_query_function_passes_tool_aliases_to_query(self):
"""InternalClient passes tool_aliases through to the Query constructor."""

async def _test():
with (
patch(
"claude_agent_sdk._internal.client.SubprocessCLITransport"
) as mock_transport_class,
patch("claude_agent_sdk._internal.client.Query") as mock_query_class,
):
mock_transport_class.return_value = self._make_mock_transport()
mock_query_class.return_value = self._make_mock_query()

options = ClaudeAgentOptions(
tool_aliases={"Read": "mcp__fakefs__Read"}
)
async for _ in query(prompt="test", options=options):
pass

call_kwargs = mock_query_class.call_args.kwargs
assert call_kwargs["tool_aliases"] == {"Read": "mcp__fakefs__Read"}

anyio.run(_test)

def test_query_function_passes_none_tool_aliases_to_query(self):
"""InternalClient passes tool_aliases=None when the option is unset."""

async def _test():
with (
patch(
"claude_agent_sdk._internal.client.SubprocessCLITransport"
) as mock_transport_class,
patch("claude_agent_sdk._internal.client.Query") as mock_query_class,
):
mock_transport_class.return_value = self._make_mock_transport()
mock_query_class.return_value = self._make_mock_query()

async for _ in query(prompt="test", options=ClaudeAgentOptions()):
pass

call_kwargs = mock_query_class.call_args.kwargs
assert call_kwargs["tool_aliases"] is None

anyio.run(_test)

def test_claude_sdk_client_passes_tool_aliases_to_query(self):
"""ClaudeSDKClient passes tool_aliases through to the Query constructor."""
from claude_agent_sdk import ClaudeSDKClient

async def _test():
with patch("claude_agent_sdk._internal.query.Query") as mock_query_class:
mock_query_class.return_value = self._make_mock_query()

options = ClaudeAgentOptions(
tool_aliases={"Bash": "mcp__workspace__bash"}
)
async with ClaudeSDKClient(
options=options, transport=self._make_mock_transport()
):
pass

call_kwargs = mock_query_class.call_args.kwargs
assert call_kwargs["tool_aliases"] == {"Bash": "mcp__workspace__bash"}

anyio.run(_test)

def test_claude_sdk_client_passes_none_tool_aliases_to_query(self):
"""ClaudeSDKClient passes tool_aliases=None when the option is unset."""
from claude_agent_sdk import ClaudeSDKClient

async def _test():
with patch("claude_agent_sdk._internal.query.Query") as mock_query_class:
mock_query_class.return_value = self._make_mock_query()

async with ClaudeSDKClient(
options=ClaudeAgentOptions(), transport=self._make_mock_transport()
):
pass

call_kwargs = mock_query_class.call_args.kwargs
assert call_kwargs["tool_aliases"] is None

anyio.run(_test)
14 changes: 14 additions & 0 deletions tests/test_query.py
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,20 @@ def test_initialize_omits_skills_for_none_and_all():
assert "skills" not in _capture_initialize_request(skills="all")


def test_initialize_sends_tool_aliases():
"""Query.initialize() includes toolAliases in the control request when set."""
sent = _capture_initialize_request(tool_aliases={"Bash": "mcp__workspace__bash"})
assert sent["subtype"] == "initialize"
assert sent["toolAliases"] == {"Bash": "mcp__workspace__bash"}


def test_initialize_omits_tool_aliases_when_unset():
"""toolAliases is absent from initialize when not configured or empty."""
assert "toolAliases" not in _capture_initialize_request()
assert "toolAliases" not in _capture_initialize_request(tool_aliases=None)
assert "toolAliases" not in _capture_initialize_request(tool_aliases={})


def _make_mock_transport(messages, control_requests=None):
"""Create a mock transport that yields messages and optionally sends control requests.

Expand Down