feat: use json schema for tool argument serialization

  - Replace Python representation with JsonSchema for tool arguments
  - Remove deprecated PydanticSchemaParser in favor of direct schema generation
  - Add handling for VAR_POSITIONAL and VAR_KEYWORD parameters
  - Improve tool argument schema collection

Author: greysonlalonde

lib/crewai/src/crewai/agent/utils.py

@@ -16,7 +16,7 @@
     KnowledgeSearchQueryFailedEvent,
 )
 from crewai.knowledge.utils.knowledge_utils import extract_knowledge_context
-from crewai.utilities.converter import generate_model_description
+from crewai.utilities.pydantic_schema_utils import generate_model_description
 
 
 if TYPE_CHECKING:

lib/crewai/src/crewai/agents/agent_adapters/base_converter_adapter.py

@@ -5,10 +5,9 @@
 from abc import ABC, abstractmethod
 import json
 import re
-from typing import TYPE_CHECKING, Final, Literal
-
-from crewai.utilities.converter import generate_model_description
+from typing import TYPE_CHECKING, Any, Final, Literal
 
+from crewai.utilities.pydantic_schema_utils import generate_model_description
 
 
 if TYPE_CHECKING:
@@ -42,7 +41,7 @@ def __init__(self, agent_adapter: BaseAgentAdapter) -> None:
         """
         self.agent_adapter = agent_adapter
         self._output_format: Literal["json", "pydantic"] | None = None
-        self._schema: str | None = None
+        self._schema: dict[str, Any] | None = None
 
     @abstractmethod
     def configure_structured_output(self, task: Task) -> None:
@@ -129,7 +128,7 @@ def _extract_json_from_text(result: str) -> str:
     @staticmethod
     def _configure_format_from_task(
         task: Task,
-    ) -> tuple[Literal["json", "pydantic"] | None, str | None]:
+    ) -> tuple[Literal["json", "pydantic"] | None, dict[str, Any] | None]:
         """Determine output format and schema from task requirements.
 
         This is a helper method that examines the task's output requirements

lib/crewai/src/crewai/agents/agent_adapters/openai_agents/structured_output_converter.py

@@ -4,6 +4,7 @@
 output conversion for OpenAI agents, supporting JSON and Pydantic model formats.
 """
 
+import json
 from typing import Any
 
 from crewai.agents.agent_adapters.base_converter_adapter import BaseConverterAdapter
@@ -61,7 +62,7 @@ def enhance_system_prompt(self, base_prompt: str) -> str:
         output_schema: str = (
             get_i18n()
             .slice("formatted_task_instructions")
-            .format(output_format=self._schema)
+            .format(output_format=json.dumps(self._schema, indent=2))
         )
 
         return f"{base_prompt}\n\n{output_schema}"

lib/crewai/src/crewai/llms/providers/azure/completion.py

@@ -9,10 +9,10 @@
 from typing_extensions import Self
 
 from crewai.utilities.agent_utils import is_context_length_exceeded
-from crewai.utilities.converter import generate_model_description
 from crewai.utilities.exceptions.context_window_exceeding_exception import (
     LLMContextLengthExceededError,
 )
+from crewai.utilities.pydantic_schema_utils import generate_model_description
 from crewai.utilities.types import LLMMessage
 
 

lib/crewai/src/crewai/llms/providers/openai/completion.py

@@ -18,10 +18,10 @@
 from crewai.llms.base_llm import BaseLLM
 from crewai.llms.hooks.transport import AsyncHTTPTransport, HTTPTransport
 from crewai.utilities.agent_utils import is_context_length_exceeded
-from crewai.utilities.converter import generate_model_description
 from crewai.utilities.exceptions.context_window_exceeding_exception import (
     LLMContextLengthExceededError,
 )
+from crewai.utilities.pydantic_schema_utils import generate_model_description
 from crewai.utilities.types import LLMMessage
 
 

lib/crewai/src/crewai/tools/base_tool.py

@@ -3,15 +3,13 @@
 from abc import ABC, abstractmethod
 import asyncio
 from collections.abc import Awaitable, Callable
-from inspect import signature
+from inspect import Parameter, signature
+import json
 from typing import (
     Any,
     Generic,
     ParamSpec,
     TypeVar,
-    cast,
-    get_args,
-    get_origin,
     overload,
 )
 
@@ -27,6 +25,7 @@
 
 from crewai.tools.structured_tool import CrewStructuredTool
 from crewai.utilities.printer import Printer
+from crewai.utilities.pydantic_schema_utils import generate_model_description
 
 
 _printer = Printer()
@@ -103,20 +102,40 @@ def _default_args_schema(
         if v != cls._ArgsSchemaPlaceholder:
             return v
 
-        return cast(
-            type[PydanticBaseModel],
-            type(
-                f"{cls.__name__}Schema",
-                (PydanticBaseModel,),
-                {
-                    "__annotations__": {
-                        k: v
-                        for k, v in cls._run.__annotations__.items()
-                        if k != "return"
-                    },
-                },
-            ),
-        )
+        run_sig = signature(cls._run)
+        fields: dict[str, Any] = {}
+
+        for param_name, param in run_sig.parameters.items():
+            if param_name in ("self", "return"):
+                continue
+            if param.kind in (Parameter.VAR_POSITIONAL, Parameter.VAR_KEYWORD):
+                continue
+
+            annotation = param.annotation if param.annotation != param.empty else Any
+
+            if param.default is param.empty:
+                fields[param_name] = (annotation, ...)
+            else:
+                fields[param_name] = (annotation, param.default)
+
+        if not fields:
+            arun_sig = signature(cls._arun)
+            for param_name, param in arun_sig.parameters.items():
+                if param_name in ("self", "return"):
+                    continue
+                if param.kind in (Parameter.VAR_POSITIONAL, Parameter.VAR_KEYWORD):
+                    continue
+
+                annotation = (
+                    param.annotation if param.annotation != param.empty else Any
+                )
+
+                if param.default is param.empty:
+                    fields[param_name] = (annotation, ...)
+                else:
+                    fields[param_name] = (annotation, param.default)
+
+        return create_model(f"{cls.__name__}Schema", **fields)
 
     @field_validator("max_usage_count", mode="before")
     @classmethod
@@ -226,24 +245,23 @@ def from_langchain(cls, tool: Any) -> BaseTool:
         args_schema = getattr(tool, "args_schema", None)
 
         if args_schema is None:
-            # Infer args_schema from the function signature if not provided
             func_signature = signature(tool.func)
-            annotations = func_signature.parameters
-            args_fields: dict[str, Any] = {}
-            for name, param in annotations.items():
-                if name != "self":
-                    param_annotation = (
-                        param.annotation if param.annotation != param.empty else Any
-                    )
-                    field_info = Field(
-                        default=...,
-                        description="",
-                    )
-                    args_fields[name] = (param_annotation, field_info)
-            if args_fields:
-                args_schema = create_model(f"{tool.name}Input", **args_fields)
+            fields: dict[str, Any] = {}
+            for name, param in func_signature.parameters.items():
+                if name == "self":
+                    continue
+                if param.kind in (Parameter.VAR_POSITIONAL, Parameter.VAR_KEYWORD):
+                    continue
+                param_annotation = (
+                    param.annotation if param.annotation != param.empty else Any
+                )
+                if param.default is param.empty:
+                    fields[name] = (param_annotation, ...)
+                else:
+                    fields[name] = (param_annotation, param.default)
+            if fields:
+                args_schema = create_model(f"{tool.name}Input", **fields)
             else:
-                # Create a default schema with no fields if no parameters are found
                 args_schema = create_model(
                     f"{tool.name}Input", __base__=PydanticBaseModel
                 )
@@ -257,53 +275,37 @@ def from_langchain(cls, tool: Any) -> BaseTool:
 
     def _set_args_schema(self) -> None:
         if self.args_schema is None:
-            class_name = f"{self.__class__.__name__}Schema"
-            self.args_schema = cast(
-                type[PydanticBaseModel],
-                type(
-                    class_name,
-                    (PydanticBaseModel,),
-                    {
-                        "__annotations__": {
-                            k: v
-                            for k, v in self._run.__annotations__.items()
-                            if k != "return"
-                        },
-                    },
-                ),
-            )
+            run_sig = signature(self._run)
+            fields: dict[str, Any] = {}
 
-    def _generate_description(self) -> None:
-        args_schema = {
-            name: {
-                "description": field.description,
-                "type": BaseTool._get_arg_annotations(field.annotation),
-            }
-            for name, field in self.args_schema.model_fields.items()
-        }
-
-        self.description = f"Tool Name: {self.name}\nTool Arguments: {args_schema}\nTool Description: {self.description}"
-
-    @staticmethod
-    def _get_arg_annotations(annotation: type[Any] | None) -> str:
-        if annotation is None:
-            return "None"
-
-        origin = get_origin(annotation)
-        args = get_args(annotation)
-
-        if origin is None:
-            return (
-                annotation.__name__
-                if hasattr(annotation, "__name__")
-                else str(annotation)
-            )
+            for param_name, param in run_sig.parameters.items():
+                if param_name in ("self", "return"):
+                    continue
+                if param.kind in (Parameter.VAR_POSITIONAL, Parameter.VAR_KEYWORD):
+                    continue
+
+                annotation = (
+                    param.annotation if param.annotation != param.empty else Any
+                )
 
-        if args:
-            args_str = ", ".join(BaseTool._get_arg_annotations(arg) for arg in args)
-            return str(f"{origin.__name__}[{args_str}]")
+                if param.default is param.empty:
+                    fields[param_name] = (annotation, ...)
+                else:
+                    fields[param_name] = (annotation, param.default)
 
-        return str(origin.__name__)
+            self.args_schema = create_model(
+                f"{self.__class__.__name__}Schema", **fields
+            )
+
+    def _generate_description(self) -> None:
+        """Generate the tool description with a JSON schema for arguments."""
+        schema = generate_model_description(self.args_schema)
+        args_json = json.dumps(schema["json_schema"]["schema"], indent=2)
+        self.description = (
+            f"Tool Name: {self.name}\n"
+            f"Tool Arguments: {args_json}\n"
+            f"Tool Description: {self.description}"
+        )
 
 
 class Tool(BaseTool, Generic[P, R]):
@@ -406,24 +408,23 @@ def from_langchain(cls, tool: Any) -> Tool[..., Any]:
         args_schema = getattr(tool, "args_schema", None)
 
         if args_schema is None:
-            # Infer args_schema from the function signature if not provided
             func_signature = signature(tool.func)
-            annotations = func_signature.parameters
-            args_fields: dict[str, Any] = {}
-            for name, param in annotations.items():
-                if name != "self":
-                    param_annotation = (
-                        param.annotation if param.annotation != param.empty else Any
-                    )
-                    field_info = Field(
-                        default=...,
-                        description="",
-                    )
-                    args_fields[name] = (param_annotation, field_info)
-            if args_fields:
-                args_schema = create_model(f"{tool.name}Input", **args_fields)
+            fields: dict[str, Any] = {}
+            for name, param in func_signature.parameters.items():
+                if name == "self":
+                    continue
+                if param.kind in (Parameter.VAR_POSITIONAL, Parameter.VAR_KEYWORD):
+                    continue
+                param_annotation = (
+                    param.annotation if param.annotation != param.empty else Any
+                )
+                if param.default is param.empty:
+                    fields[name] = (param_annotation, ...)
+                else:
+                    fields[name] = (param_annotation, param.default)
+            if fields:
+                args_schema = create_model(f"{tool.name}Input", **fields)
             else:
-                # Create a default schema with no fields if no parameters are found
                 args_schema = create_model(
                     f"{tool.name}Input", __base__=PydanticBaseModel
                 )
@@ -502,32 +503,38 @@ def _make_with_name(tool_name: str) -> Callable[[Callable[P2, R2]], Tool[P2, R2]
         def _make_tool(f: Callable[P2, R2]) -> Tool[P2, R2]:
             if f.__doc__ is None:
                 raise ValueError("Function must have a docstring")
-
-            func_annotations = getattr(f, "__annotations__", None)
-            if func_annotations is None:
+            if f.__annotations__ is None:
                 raise ValueError("Function must have type annotations")
 
+            func_sig = signature(f)
+            fields: dict[str, Any] = {}
+
+            for param_name, param in func_sig.parameters.items():
+                if param_name == "return":
+                    continue
+                if param.kind in (Parameter.VAR_POSITIONAL, Parameter.VAR_KEYWORD):
+                    continue
+
+                annotation = (
+                    param.annotation if param.annotation != param.empty else Any
+                )
+
+                if param.default is param.empty:
+                    fields[param_name] = (annotation, ...)
+                else:
+                    fields[param_name] = (annotation, param.default)
+
             class_name = "".join(tool_name.split()).title()
-            tool_args_schema = cast(
-                type[PydanticBaseModel],
-                type(
-                    class_name,
-                    (PydanticBaseModel,),
-                    {
-                        "__annotations__": {
-                            k: v for k, v in func_annotations.items() if k != "return"
-                        },
-                    },
-                ),
-            )
+            args_schema = create_model(class_name, **fields)
 
             return Tool(
                 name=tool_name,
                 description=f.__doc__,
                 func=f,
-                args_schema=tool_args_schema,
+                args_schema=args_schema,
                 result_as_answer=result_as_answer,
                 max_usage_count=max_usage_count,
+                current_usage_count=0,
             )
 
         return _make_tool