fix: run tool calls concurrently in the Python client

CHANGELOG.md

@@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- Moved full MCP tool call / callback inputs and outputs, and `execute_typescript`/`execute_bash` code and output, from INFO to DEBUG logs to declutter traces. Added DEBUG logs describing tool result shape (structured content, JSON parse success/fallback).
 - **Breaking:** `CodeMode::with_callbacks` returns `(Self, CallbackReport)` instead of `Result<Self>`, so builder-style callers see which tools failed or degraded. Per-tool isolation means the batch itself cannot fail, so the report is the only outcome.
 - **Breaking:** `CodeMode::add_callback` returns `Result<Vec<String>>` — the reasons that tool's types were degraded to `any`, empty when fully typed.
 

crates/pctx_code_mode/src/code_mode.rs

@@ -513,7 +513,7 @@ impl CodeMode {
     }
 
     /// Execute bash commands directly in the virtual filesystem
-    #[instrument(skip(self), ret(Display), err)]
+    #[instrument(skip(self), err)]
     pub async fn execute_bash(&self, command: &str) -> Result<ExecuteBashOutput> {
         debug!(command = %command, "Executing bash command");
 
@@ -578,15 +578,18 @@ export default result;"#,
             warn!("Bash execution failed with exit code {exit_code}: {stderr}");
         }
 
-        Ok(ExecuteBashOutput {
+        let output = ExecuteBashOutput {
             exit_code,
             stdout,
             stderr,
-        })
+        };
+        debug!("Bash execution result:\n{output}");
+
+        Ok(output)
     }
 
     /// Execute TypeScript code with access to registered tools and virtual filesystem
-    #[instrument(skip(self, registry), ret(Display), err)]
+    #[instrument(skip(self, registry, code), err)]
     pub async fn execute_typescript(
         &self,
         code: &str,
@@ -698,7 +701,7 @@ export default result;"#,
             }
         };
 
-        debug!(to_execute = %to_execute, "Executing TypeScript in sandbox");
+        debug!("Executing TypeScript in sandbox:\n{to_execute}");
 
         let execution_res = pctx_executor::execute(
             &to_execute,
@@ -712,14 +715,18 @@ export default result;"#,
             warn!("TypeScript execution failed: {:?}", execution_res.stderr);
         }
 
-        Ok(ExecuteTypescriptOutput {
+        let output = ExecuteTypescriptOutput {
             success: execution_res.success,
             stdout: execution_res.stdout,
             stderr: execution_res.stderr,
             output: execution_res.output,
             registry: execution_res.registry,
             trace: execution_res.trace,
-        })
+        };
+
+        debug!("TypeScript execution result:\n{output}");
+
+        Ok(output)
     }
 }
 

crates/pctx_registry/src/registry.rs

@@ -15,7 +15,7 @@ use std::{
     sync::{Arc, RwLock},
     time::SystemTime,
 };
-use tracing::{debug, info, instrument, warn};
+use tracing::{debug, instrument, warn};
 
 pub type CallbackFn = Arc<
     dyn Fn(
@@ -237,13 +237,7 @@ impl PctxRegistry {
     ///
     /// This function will return an error if an action by the provided id doesn't exist
     /// or if the action itself fails
-    #[instrument(
-        name = "invoke_registry_action",
-        skip_all,
-        fields(id=id, args = json!(args).to_string()),
-        ret(Display),
-        err
-    )]
+    #[instrument(name = "invoke_registry_action", skip_all, fields(id = id), err)]
     pub async fn invoke(
         &self,
         id: &str,
@@ -333,25 +327,40 @@ impl PctxRegistry {
                     }
 
                     // Prefer structuredContent if available, otherwise use content array
-                    let has_structured = tool_result.structured_content.is_some();
                     let val = if let Some(structured) = tool_result.structured_content {
+                        debug!(tool = %mcp_id.id(), "tool result: using structured content");
                         structured
                     } else if let Some(RawContent::Text(text_content)) =
                         tool_result.content.first().map(|a| &**a)
                     {
                         // Try to parse as JSON, fallback to string value
-                        serde_json::from_str(&text_content.text)
-                            .or_else(|_| Ok(serde_json::Value::String(text_content.text.clone())))
-                            .map_err(|e: serde_json::Error| {
-                                RegistryError::ToolCall(format!("Failed to parse content: {e}"))
-                            })?
+                        match serde_json::from_str(&text_content.text) {
+                            Ok(json) => {
+                                debug!(
+                                    tool = %mcp_id.id(),
+                                    "tool result: parsed text content as JSON"
+                                );
+                                json
+                            }
+                            Err(e) => {
+                                debug!(
+                                    tool = %mcp_id.id(),
+                                    error = %e,
+                                    "tool result: text content is not JSON, using raw string"
+                                );
+                                serde_json::Value::String(text_content.text.clone())
+                            }
+                        }
                     } else {
                         // Return the whole content array as JSON
+                        debug!(
+                            tool = %mcp_id.id(),
+                            content_len = tool_result.content.len(),
+                            "tool result: no structured or text content, using raw content array"
+                        );
                         json!(tool_result.content)
                     };
 
-                    info!(structured_content = has_structured, result =? &val, "Tool result");
-
                     Ok(val)
                 })();
 

pctx-py/CHANGELOG.md

@@ -17,6 +17,14 @@ For changes to the underlying Rust crates and CLI, see the
 
 ### Fixed
 
+- Tool calls now run concurrently. Each request is handled in its own task
+  rather than awaited inside the WebSocket read loop, so code fanning out with
+  `Promise.all` takes the time of its slowest call instead of the sum of all
+  of them.
+- Sync tools run on a worker thread instead of blocking the event loop, so one
+  slow sync tool no longer stalls the calls beside it. Their bodies now execute
+  off the main thread, so anything they share must be thread-safe.
+
 ## [v0.4.4] - 2026-07-22
 
 ### Fixed

pctx-py/src/pctx_client/_websocket_client.py

@@ -68,6 +68,10 @@ def __init__(
         self._headers = headers or {}
         self._pending_executions: dict[str | int, asyncio.Future] = {}
         self._request_counter = 0
+        self._message_handler_task: asyncio.Task | None = None
+        # In-flight tool executions. Held strongly because asyncio only keeps
+        # weak references to tasks, and cancelled as a group on disconnect.
+        self._tool_tasks: set[asyncio.Task] = set()
 
     async def _connect(self, code_mode_session: str):
         """
@@ -100,6 +104,10 @@ async def _disconnect(self):
         if self._message_handler_task:
             self._message_handler_task.cancel()
 
+        for task in self._tool_tasks:
+            task.cancel()
+        self._tool_tasks.clear()
+
         if self.ws:
             await self.ws.close()
             self.ws = None
@@ -192,8 +200,14 @@ async def _handle_messages(self):
                     message: WebSocketMessage = adapter.validate_json(message_data)
 
                     if isinstance(message, ExecuteToolRequest):
-                        res = await self._handle_execute_tool(message)
-                        await self._send(res)
+                        # Run the tool in its own task so this loop stays free to
+                        # read the next message. Awaiting it here would serialize
+                        # every tool call the server dispatches, so code that fans
+                        # out with `Promise.all` would take the sum of its calls
+                        # rather than the slowest one.
+                        task = asyncio.create_task(self._execute_tool(message))
+                        self._tool_tasks.add(task)
+                        task.add_done_callback(self._tool_tasks.discard)
                     elif isinstance(message, ExecuteCodeResponse):
                         future = self._pending_executions.get(message.id)
                         if future is not None:
@@ -216,6 +230,16 @@ async def _handle_messages(self):
         except Exception as e:
             print(f"Message handler error: {e}")
 
+    async def _execute_tool(self, req: ExecuteToolRequest):
+        """Run one tool request and send its response back to the server."""
+        try:
+            res = await self._handle_execute_tool(req)
+            await self._send(res)
+        except asyncio.CancelledError:
+            raise
+        except Exception as e:
+            print(f"Error executing tool: {e}")
+
     async def _handle_execute_tool(
         self, req: ExecuteToolRequest
     ) -> ExecuteToolResponse | JsonRpcError:
@@ -240,10 +264,14 @@ async def _handle_execute_tool(
         args = req.params.args or {}
         try:
             if isinstance(tool, Tool):
+                # Sync tools go to a worker thread. Calling one inline would
+                # block the event loop for its whole duration, which stalls
+                # every other in-flight tool call behind it -- and with it the
+                # loop reading further requests off the WebSocket.
                 if tool.input_schema is None:
-                    output = tool.invoke()
+                    output = await asyncio.to_thread(tool.invoke)
                 else:
-                    output = tool.invoke(**args)
+                    output = await asyncio.to_thread(tool.invoke, **args)
             else:
                 if tool.input_schema is None:
                     output = await tool.ainvoke()

pctx-py/tests/test_integration.py

@@ -1,5 +1,8 @@
 """Integration tests for pctx code mode against a running server"""
 
+import asyncio
+import threading
+import time
 from datetime import datetime
 
 import pytest
@@ -796,3 +799,150 @@ def format_result(value: int, label: str) -> str:
             "Please ensure the pctx server is running.\n"
             "Start the server with: pctx server start"
         )
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_concurrent_async_tool_calls_run_in_parallel():
+    """Tools fanned out with `Promise.all` must execute concurrently.
+
+    The client used to await each tool request inside its WebSocket read loop,
+    so a batch of N calls took the sum of their durations instead of the
+    slowest one. Assert on observed overlap rather than wall time alone, so
+    the test fails on serialization rather than on a slow machine.
+    """
+    sleep_secs = 0.5
+    calls = 4
+
+    in_flight = 0
+    max_in_flight = 0
+
+    @tool
+    async def slow_echo(value: int) -> int:
+        """Sleep briefly, then echo the value back"""
+        nonlocal in_flight, max_in_flight
+        in_flight += 1
+        max_in_flight = max(max_in_flight, in_flight)
+        try:
+            await asyncio.sleep(sleep_secs)
+            return value
+        finally:
+            in_flight -= 1
+
+    try:
+        async with Pctx(tools=[slow_echo], execute_timeout=60) as pctx:
+            code = """
+            async function run() {
+                const values = await Promise.all([
+                    Tools.slowEcho({ value: 1 }),
+                    Tools.slowEcho({ value: 2 }),
+                    Tools.slowEcho({ value: 3 }),
+                    Tools.slowEcho({ value: 4 }),
+                ]);
+                return { values };
+            }
+            """
+
+            start = time.perf_counter()
+            output = await pctx.execute_typescript(code)
+            elapsed = time.perf_counter() - start
+
+            assert output.success, f"Execution should succeed, got: {output.stderr}"
+            assert output.output is not None, "Execution should return output"
+            assert output.output.get("values") == [1, 2, 3, 4], (
+                f"Expected all four calls to return, got: {output.output}"
+            )
+
+            assert max_in_flight == calls, (
+                f"All {calls} tool calls should be in flight at once, "
+                f"peaked at {max_in_flight} -- the client is serializing them"
+            )
+
+            # Serialized dispatch takes calls * sleep_secs; concurrent dispatch
+            # takes ~sleep_secs. Half way between the two is a wide enough
+            # margin to absorb session setup and round-trip overhead.
+            serial_secs = calls * sleep_secs
+            assert elapsed < serial_secs / 2, (
+                f"Concurrent calls took {elapsed:.2f}s; serialized dispatch "
+                f"would take ~{serial_secs:.2f}s"
+            )
+    except ConnectionError:
+        pytest.fail(
+            "Failed to connect to pctx server at http://localhost:8080.\n"
+            "Please ensure the pctx server is running.\n"
+            "Start the server with: pctx server start"
+        )
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_concurrent_sync_tool_calls_run_in_parallel():
+    """Sync tools fanned out with `Promise.all` must also run concurrently.
+
+    A sync tool body blocks whatever thread it runs on, so calling it inline
+    on the event loop would stall every other in-flight call behind it. They
+    run on worker threads instead -- hence the blocking `time.sleep` here, and
+    the lock around the counters, which the tool bodies touch off-thread.
+    """
+    sleep_secs = 0.5
+    calls = 4
+
+    lock = threading.Lock()
+    in_flight = 0
+    max_in_flight = 0
+
+    @tool
+    def slow_echo_sync(value: int) -> int:
+        """Block briefly, then echo the value back"""
+        nonlocal in_flight, max_in_flight
+        with lock:
+            in_flight += 1
+            max_in_flight = max(max_in_flight, in_flight)
+        try:
+            time.sleep(sleep_secs)
+            return value
+        finally:
+            with lock:
+                in_flight -= 1
+
+    try:
+        async with Pctx(tools=[slow_echo_sync], execute_timeout=60) as pctx:
+            code = """
+            async function run() {
+                const values = await Promise.all([
+                    Tools.slowEchoSync({ value: 1 }),
+                    Tools.slowEchoSync({ value: 2 }),
+                    Tools.slowEchoSync({ value: 3 }),
+                    Tools.slowEchoSync({ value: 4 }),
+                ]);
+                return { values };
+            }
+            """
+
+            start = time.perf_counter()
+            output = await pctx.execute_typescript(code)
+            elapsed = time.perf_counter() - start
+
+            assert output.success, f"Execution should succeed, got: {output.stderr}"
+            assert output.output is not None, "Execution should return output"
+            assert output.output.get("values") == [1, 2, 3, 4], (
+                f"Expected all four calls to return, got: {output.output}"
+            )
+
+            assert max_in_flight == calls, (
+                f"All {calls} tool calls should be in flight at once, "
+                f"peaked at {max_in_flight} -- sync tools are blocking the "
+                f"event loop instead of running on worker threads"
+            )
+
+            serial_secs = calls * sleep_secs
+            assert elapsed < serial_secs / 2, (
+                f"Concurrent calls took {elapsed:.2f}s; serialized dispatch "
+                f"would take ~{serial_secs:.2f}s"
+            )
+    except ConnectionError:
+        pytest.fail(
+            "Failed to connect to pctx server at http://localhost:8080.\n"
+            "Please ensure the pctx server is running.\n"
+            "Start the server with: pctx server start"
+        )