fix: send agent definitions via initialize request matching TypeScript SDK

e2e-tests/test_agents_and_settings.py

@@ -1,6 +1,7 @@
 """End-to-end tests for agents and setting sources with real Claude API calls."""
 
 import asyncio
+import json
 import sys
 import tempfile
 from pathlib import Path
@@ -15,10 +16,33 @@
 )
 
 
+def generate_large_agents(
+    num_agents: int = 20, prompt_size_kb: int = 12
+) -> dict[str, AgentDefinition]:
+    """Generate multiple agents with large prompts for testing.
+
+    Args:
+        num_agents: Number of agents to generate
+        prompt_size_kb: Size of each agent's prompt in KB
+
+    Returns:
+        Dictionary of agent name -> AgentDefinition
+    """
+    agents = {}
+    for i in range(num_agents):
+        # Generate a large prompt with some structure
+        prompt_content = f"You are test agent #{i}. " + ("x" * (prompt_size_kb * 1024))
+        agents[f"large-agent-{i}"] = AgentDefinition(
+            description=f"Large test agent #{i} for stress testing",
+            prompt=prompt_content,
+        )
+    return agents
+
+
 @pytest.mark.e2e
 @pytest.mark.asyncio
 async def test_agent_definition():
-    """Test that custom agent definitions work."""
+    """Test that custom agent definitions work in streaming mode."""
     options = ClaudeAgentOptions(
         agents={
             "test-agent": AgentDefinition(
@@ -47,6 +71,74 @@ async def test_agent_definition():
                 break
 
 
+@pytest.mark.e2e
+@pytest.mark.asyncio
+async def test_agent_definition_with_query_function():
+    """Test that custom agent definitions work with the query() function.
+
+    Both ClaudeSDKClient and query() now use streaming mode internally,
+    sending agents via the initialize request.
+    """
+    from claude_agent_sdk import query
+
+    options = ClaudeAgentOptions(
+        agents={
+            "test-agent-query": AgentDefinition(
+                description="A test agent for query function verification",
+                prompt="You are a test agent.",
+            )
+        },
+        max_turns=1,
+    )
+
+    # Use query() with string prompt
+    found_agent = False
+    async for message in query(prompt="What is 2 + 2?", options=options):
+        if isinstance(message, SystemMessage) and message.subtype == "init":
+            agents = message.data.get("agents", [])
+            assert "test-agent-query" in agents, (
+                f"test-agent-query should be available, got: {agents}"
+            )
+            found_agent = True
+            break
+
+    assert found_agent, "Should have received init message with agents"
+
+
+@pytest.mark.e2e
+@pytest.mark.asyncio
+async def test_large_agents_with_query_function():
+    """Test large agent definitions (260KB+) work with query() function.
+
+    Since we now always use streaming mode internally (matching TypeScript SDK),
+    large agents are sent via the initialize request through stdin with no
+    size limits.
+    """
+    from claude_agent_sdk import query
+
+    # Generate 20 agents with 13KB prompts each = ~260KB total
+    agents = generate_large_agents(num_agents=20, prompt_size_kb=13)
+
+    options = ClaudeAgentOptions(
+        agents=agents,
+        max_turns=1,
+    )
+
+    # Use query() with string prompt - agents still go via initialize
+    found_agents = []
+    async for message in query(prompt="What is 2 + 2?", options=options):
+        if isinstance(message, SystemMessage) and message.subtype == "init":
+            found_agents = message.data.get("agents", [])
+            break
+
+    # Check all our agents are registered
+    for agent_name in agents:
+        assert agent_name in found_agents, (
+            f"{agent_name} should be registered. "
+            f"Found: {found_agents[:5]}... ({len(found_agents)} total)"
+        )
+
+
 @pytest.mark.e2e
 @pytest.mark.asyncio
 async def test_filesystem_agent_loading():
@@ -240,3 +332,62 @@ async def test_setting_sources_project_included():
         # On Windows, wait for file handles to be released before cleanup
         if sys.platform == "win32":
             await asyncio.sleep(0.5)
+
+
+@pytest.mark.e2e
+@pytest.mark.asyncio
+async def test_large_agent_definitions_via_initialize():
+    """Test that large agent definitions (250KB+) are sent via initialize request.
+
+    This test verifies the fix for the issue where large agent definitions
+    would previously trigger a temp file workaround with @filepath. Now they
+    are sent via the initialize control request through stdin, which has no
+    size limit.
+
+    The test:
+    1. Generates 20 agents with ~13KB prompts each (~260KB total)
+    2. Creates an SDK client with these agents
+    3. Verifies all agents are registered and available
+    """
+    from dataclasses import asdict
+
+    # Generate 20 agents with 13KB prompts each = ~260KB total
+    agents = generate_large_agents(num_agents=20, prompt_size_kb=13)
+
+    # Calculate total size to verify we're testing the right thing
+    total_size = sum(
+        len(json.dumps({k: v for k, v in asdict(agent).items() if v is not None}))
+        for agent in agents.values()
+    )
+    assert total_size > 250_000, (
+        f"Test agents should be >250KB, got {total_size / 1024:.1f}KB"
+    )
+
+    options = ClaudeAgentOptions(
+        agents=agents,
+        max_turns=1,
+    )
+
+    async with ClaudeSDKClient(options=options) as client:
+        await client.query("List available agents")
+
+        # Check that all agents are available in init message
+        async for message in client.receive_response():
+            if isinstance(message, SystemMessage) and message.subtype == "init":
+                registered_agents = message.data.get("agents", [])
+                assert isinstance(registered_agents, list), (
+                    f"agents should be a list, got: {type(registered_agents)}"
+                )
+
+                # Verify all our agents are registered
+                for agent_name in agents:
+                    assert agent_name in registered_agents, (
+                        f"{agent_name} should be registered. "
+                        f"Found: {registered_agents[:5]}... ({len(registered_agents)} total)"
+                    )
+
+                # All agents should be there
+                assert len(registered_agents) >= len(agents), (
+                    f"Expected at least {len(agents)} agents, got {len(registered_agents)}"
+                )
+                break

src/claude_agent_sdk/_internal/client.py

@@ -1,7 +1,7 @@
 """Internal client implementation."""
 
 from collections.abc import AsyncIterable, AsyncIterator
-from dataclasses import replace
+from dataclasses import asdict, replace
 from typing import Any
 
 from ..types import (
@@ -89,32 +89,52 @@ async def process_query(
                 if isinstance(config, dict) and config.get("type") == "sdk":
                     sdk_mcp_servers[name] = config["instance"]  # type: ignore[typeddict-item]
 
+        # Convert agents to dict format for initialize request
+        agents_dict = None
+        if configured_options.agents:
+            agents_dict = {
+                name: {k: v for k, v in asdict(agent_def).items() if v is not None}
+                for name, agent_def in configured_options.agents.items()
+            }
+
         # Create Query to handle control protocol
-        is_streaming = not isinstance(prompt, str)
+        # Always use streaming mode internally (matching TypeScript SDK)
+        # This ensures agents are always sent via initialize request
         query = Query(
             transport=chosen_transport,
-            is_streaming_mode=is_streaming,
+            is_streaming_mode=True,  # Always streaming internally
             can_use_tool=configured_options.can_use_tool,
             hooks=self._convert_hooks_to_internal_format(configured_options.hooks)
             if configured_options.hooks
             else None,
             sdk_mcp_servers=sdk_mcp_servers,
+            agents=agents_dict,
         )
 
         try:
             # Start reading messages
             await query.start()
 
-            # Initialize if streaming
-            if is_streaming:
-                await query.initialize()
+            # Always initialize to send agents via stdin (matching TypeScript SDK)
+            await query.initialize()
 
-            # Stream input if it's an AsyncIterable
-            if isinstance(prompt, AsyncIterable) and query._tg:
-                # Start streaming in background
-                # Create a task that will run in the background
+            # Handle prompt input
+            if isinstance(prompt, str):
+                # For string prompts, write user message to stdin after initialize
+                # (matching TypeScript SDK behavior)
+                import json
+
+                user_message = {
+                    "type": "user",
+                    "session_id": "",
+                    "message": {"role": "user", "content": prompt},
+                    "parent_tool_use_id": None,
+                }
+                await chosen_transport.write(json.dumps(user_message) + "\n")
+                await chosen_transport.end_input()
+            elif isinstance(prompt, AsyncIterable) and query._tg:
+                # Stream input in background for async iterables
                 query._tg.start_soon(query.stream_input, prompt)
-            # For string prompts, the prompt is already passed via CLI args
 
             # Yield parsed messages
             async for data in query.receive_messages():

src/claude_agent_sdk/_internal/query.py

@@ -73,6 +73,7 @@ def __init__(
         hooks: dict[str, list[dict[str, Any]]] | None = None,
         sdk_mcp_servers: dict[str, "McpServer"] | None = None,
         initialize_timeout: float = 60.0,
+        agents: dict[str, dict[str, Any]] | None = None,
     ):
         """Initialize Query with transport and callbacks.
 
@@ -83,13 +84,15 @@ def __init__(
             hooks: Optional hook configurations
             sdk_mcp_servers: Optional SDK MCP server instances
             initialize_timeout: Timeout in seconds for the initialize request
+            agents: Optional agent definitions to send via initialize
         """
         self._initialize_timeout = initialize_timeout
         self.transport = transport
         self.is_streaming_mode = is_streaming_mode
         self.can_use_tool = can_use_tool
         self.hooks = hooks or {}
         self.sdk_mcp_servers = sdk_mcp_servers or {}
+        self._agents = agents
 
         # Control protocol state
         self.pending_control_responses: dict[str, anyio.Event] = {}
@@ -144,10 +147,12 @@ async def initialize(self) -> dict[str, Any] | None:
                         hooks_config[event].append(hook_matcher_config)
 
         # Send initialize request
-        request = {
+        request: dict[str, Any] = {
             "subtype": "initialize",
             "hooks": hooks_config if hooks_config else None,
         }
+        if self._agents:
+            request["agents"] = self._agents
 
         # Use longer timeout for initialize since MCP servers may take time to start
         response = await self._send_control_request(