Python: sanitize handoff tool names for agent names with spaces

python/packages/orchestrations/agent_framework_orchestrations/_handoff.py

@@ -32,6 +32,7 @@
 import inspect
 import json
 import logging
+import re
 import sys
 from collections.abc import Awaitable, Callable, Mapping, Sequence
 from copy import deepcopy
@@ -124,7 +125,8 @@ def __hash__(self) -> int:
 
 def get_handoff_tool_name(target_id: str) -> str:
     """Get the standardized handoff tool name for a given target agent ID."""
-    return f"handoff_to_{target_id}"
+    sanitized_target_id = re.sub(r"[^a-zA-Z0-9_-]", "_", target_id)
+    return f"handoff_to_{sanitized_target_id}"
 
 
 HANDOFF_FUNCTION_RESULT_KEY = "handoff_to"
@@ -134,9 +136,17 @@ class _AutoHandoffMiddleware(FunctionMiddleware):
     """Intercept handoff tool invocations and short-circuit execution with synthetic results."""
 
     def __init__(self, handoffs: Sequence[HandoffConfiguration]) -> None:
-        """Initialise middleware with the mapping from tool name to specialist id."""
-        self._handoff_functions = {get_handoff_tool_name(handoff.target_id): handoff.target_id for handoff in handoffs}
         self._invoked_handoffs: list[tuple[str, str]] = []
+        self._handoff_functions: dict[str, str] = {}
+        for handoff in handoffs:
+            tool_name = get_handoff_tool_name(handoff.target_id)
+            existing_target = self._handoff_functions.get(tool_name)
+            if existing_target is not None and existing_target != handoff.target_id:
+                raise ValueError(
+                    f"Handoff targets {existing_target!r} and {handoff.target_id!r} "
+                    f"produce the same tool name {tool_name!r}."
+                )
+            self._handoff_functions[tool_name] = handoff.target_id
 
     async def process(
         self,

python/packages/orchestrations/tests/test_handoff.py

@@ -1535,6 +1535,47 @@ def test_handoff_builder_accepts_all_instances_in_add_handoff():
     assert "specialist_b" in workflow.executors
 
 
+def test_get_handoff_tool_name_sanitizes_target_id() -> None:
+    """Handoff tool names should be valid when target IDs contain spaces or punctuation."""
+    assert get_handoff_tool_name("Billing Agent") == "handoff_to_Billing_Agent"
+    assert get_handoff_tool_name("Order-Support.Agent") == "handoff_to_Order-Support_Agent"
+
+
+def test_auto_handoff_middleware_rejects_sanitized_name_collisions() -> None:
+    """Middleware should reject targets that sanitize to the same handoff tool name."""
+    with pytest.raises(ValueError, match="produce the same tool name"):
+        _AutoHandoffMiddleware([
+            HandoffConfiguration(target="Billing Agent"),
+            HandoffConfiguration(target="Billing.Agent"),
+        ])
+
+
+async def test_auto_handoff_with_sanitized_target_id() -> None:
+    """Handoff tools should sanitize names while preserving the original target ID for routing."""
+    target_id = "Billing Agent"
+    executor = HandoffAgentExecutor(
+        MockHandoffAgent(name="triage"),
+        [HandoffConfiguration(target=target_id)],
+    )
+    executor_agent = _as_handoff_agent(executor._agent)  # pyright: ignore[reportPrivateUsage]
+
+    handoff_tool = next(
+        tool
+        for tool in executor_agent.default_options["tools"]
+        if isinstance(tool, FunctionTool) and tool.name == "handoff_to_Billing_Agent"
+    )
+
+    context = FunctionInvocationContext(function=handoff_tool, arguments={})
+    context.metadata["call_id"] = "handoff-call"
+
+    with pytest.raises(MiddlewareTermination):
+        await executor._auto_handoff_middleware.process(context, AsyncMock())  # pyright: ignore[reportPrivateUsage]
+
+    assert executor._auto_handoff_middleware.consume_invoked_handoffs() == [  # pyright: ignore[reportPrivateUsage]
+        ("handoff-call", target_id)
+    ]
+
+
 async def test_auto_handoff_middleware_intercepts_handoff_tool_call() -> None:
     """Middleware should short-circuit matching handoff tool calls with a synthetic result."""
     target_id = "specialist"