fix(core): support positional-only function tool args

python/packages/autogen-core/src/autogen_core/tools/_function_tool.py

@@ -1,5 +1,6 @@
 import asyncio
 import functools
+import inspect
 import warnings
 from textwrap import dedent
 from typing import Any, Callable, Sequence
@@ -103,31 +104,31 @@ def __init__(
         super().__init__(args_model, return_type, func_name, description, strict)
 
     async def run(self, args: BaseModel, cancellation_token: CancellationToken) -> Any:
-        kwargs = {}
+        positional_args: list[Any] = []
+        kwargs: dict[str, Any] = {}
+
+        for name, parameter in self._signature.parameters.items():
+            if name == "cancellation_token":
+                value = cancellation_token
+            elif hasattr(args, name):
+                value = getattr(args, name)
+            else:
+                continue
 
-        for name in self._signature.parameters.keys():
-            if hasattr(args, name):
-                kwargs[name] = getattr(args, name)
+            if parameter.kind == inspect.Parameter.POSITIONAL_ONLY:
+                positional_args.append(value)
+            else:
+                kwargs[name] = value
 
         if asyncio.iscoroutinefunction(self._func):
-            if self._has_cancellation_support:
-                result = await self._func(**kwargs, cancellation_token=cancellation_token)
-            else:
-                result = await self._func(**kwargs)
+            result = await self._func(*positional_args, **kwargs)
         else:
-            if self._has_cancellation_support:
-                result = await asyncio.get_event_loop().run_in_executor(
-                    None,
-                    functools.partial(
-                        self._func,
-                        **kwargs,
-                        cancellation_token=cancellation_token,
-                    ),
-                )
-            else:
-                future = asyncio.get_event_loop().run_in_executor(None, functools.partial(self._func, **kwargs))
+            future = asyncio.get_event_loop().run_in_executor(
+                None, functools.partial(self._func, *positional_args, **kwargs)
+            )
+            if not self._has_cancellation_support:
                 cancellation_token.link_future(future)
-                result = await future
+            result = await future
 
         return result
 

python/packages/autogen-core/tests/test_tools.py

@@ -421,6 +421,58 @@ def my_function(arg: str) -> MyResult:
     assert result.result == "value"
 
 
+@pytest.mark.asyncio
+async def test_func_tool_positional_only_argument() -> None:
+    def scale(value: int, /, *, factor: int = 2) -> int:
+        return value * factor
+
+    tool = FunctionTool(scale, description="Scale a value.")
+    result = await tool.run_json({"value": 3, "factor": 4}, CancellationToken())
+    assert result == 12
+
+
+@pytest.mark.asyncio
+async def test_func_tool_async_positional_only_argument() -> None:
+    async def scale(value: int, /, *, factor: int = 2) -> int:
+        return value * factor
+
+    tool = FunctionTool(scale, description="Scale a value.")
+    result = await tool.run_json({"value": 3, "factor": 4}, CancellationToken())
+    assert result == 12
+
+
+@pytest.mark.asyncio
+async def test_func_tool_positional_only_default_before_required_argument() -> None:
+    def combine(prefix: str = "default", /, suffix: str = "!") -> str:
+        return prefix + suffix
+
+    tool = FunctionTool(combine, description="Combine two strings.")
+    result = await tool.run_json({"suffix": "?"}, CancellationToken())
+    assert result == "default?"
+
+
+@pytest.mark.asyncio
+async def test_func_tool_positional_only_cancellation_token() -> None:
+    def use_token(value: str, cancellation_token: CancellationToken, /) -> bool:
+        return value == "ok" and cancellation_token is token
+
+    token = CancellationToken()
+    tool = FunctionTool(use_token, description="Use a cancellation token.")
+    result = await tool.run_json({"value": "ok"}, token)
+    assert result is True
+
+
+@pytest.mark.asyncio
+async def test_func_tool_async_positional_only_cancellation_token() -> None:
+    async def use_token(value: str, cancellation_token: CancellationToken, /) -> bool:
+        return value == "ok" and cancellation_token is token
+
+    token = CancellationToken()
+    tool = FunctionTool(use_token, description="Use a cancellation token.")
+    result = await tool.run_json({"value": "ok"}, token)
+    assert result is True
+
+
 @pytest.mark.asyncio
 async def test_func_str_res() -> None:
     def my_function(arg: str) -> str: