Python: [BREAKING] Remove FunctionTool[Any] compatibility shim for schema passthrough (#3600)

python/packages/ag-ui/agent_framework_ag_ui/_client.py

@@ -267,7 +267,7 @@ def _register_server_tool_placeholder(self, tool_name: str) -> None:
         if any(getattr(tool, "name", None) == tool_name for tool in additional_tools):
             return
 
-        placeholder: FunctionTool[Any] = FunctionTool(
+        placeholder: FunctionTool = FunctionTool(
             name=tool_name,
             description="Server-managed tool placeholder (AG-UI)",
             func=None,

python/packages/ag-ui/agent_framework_ag_ui/_utils.py

@@ -162,7 +162,7 @@ def make_json_safe(obj: Any) -> Any:  # noqa: ANN401
 
 def convert_agui_tools_to_agent_framework(
     agui_tools: list[dict[str, Any]] | None,
-) -> list[FunctionTool[Any]] | None:
+) -> list[FunctionTool] | None:
     """Convert AG-UI tool definitions to Agent Framework FunctionTool declarations.
 
     Creates declaration-only FunctionTool instances (no executable implementation).
@@ -181,13 +181,13 @@ def convert_agui_tools_to_agent_framework(
     if not agui_tools:
         return None
 
-    result: list[FunctionTool[Any]] = []
+    result: list[FunctionTool] = []
     for tool_def in agui_tools:
         # Create declaration-only FunctionTool (func=None means no implementation)
         # When func=None, the declaration_only property returns True,
         # which tells the function invocation mixin to return the function call
         # without executing it (so it can be sent back to the client)
-        func: FunctionTool[Any] = FunctionTool(
+        func: FunctionTool = FunctionTool(
             name=tool_def.get("name", ""),
             description=tool_def.get("description", ""),
             func=None,  # CRITICAL: Makes declaration_only=True

python/packages/ag-ui/agent_framework_ag_ui_examples/agents/ui_generator_agent.py

@@ -5,7 +5,7 @@
 from __future__ import annotations
 
 import sys
-from typing import TYPE_CHECKING, Any, TypedDict
+from typing import TYPE_CHECKING, TypedDict
 
 from agent_framework import Agent, FunctionTool, SupportsChatGetResponse
 from agent_framework.ag_ui import AgentFrameworkAgent
@@ -23,7 +23,7 @@
     from agent_framework import ChatOptions
 
 # Declaration-only tools (func=None) - actual rendering happens on the client side
-generate_haiku = FunctionTool[Any](
+generate_haiku = FunctionTool(
     name="generate_haiku",
     description="""Generate a haiku with image and gradient background (FRONTEND_RENDER).
 
@@ -71,7 +71,7 @@
     },
 )
 
-create_chart = FunctionTool[Any](
+create_chart = FunctionTool(
     name="create_chart",
     description="""Create an interactive chart (FRONTEND_RENDER).
 
@@ -99,7 +99,7 @@
     },
 )
 
-display_timeline = FunctionTool[Any](
+display_timeline = FunctionTool(
     name="display_timeline",
     description="""Display an interactive timeline (FRONTEND_RENDER).
 
@@ -127,7 +127,7 @@
     },
 )
 
-show_comparison_table = FunctionTool[Any](
+show_comparison_table = FunctionTool(
     name="show_comparison_table",
     description="""Show a comparison table (FRONTEND_RENDER).
 

python/packages/claude/agent_framework_claude/_agent.py

@@ -484,7 +484,7 @@ def _prepare_tools(
 
         return create_sdk_mcp_server(name=TOOLS_MCP_SERVER_NAME, tools=sdk_tools), tool_names
 
-    def _function_tool_to_sdk_mcp_tool(self, func_tool: FunctionTool[Any]) -> SdkMcpTool[Any]:
+    def _function_tool_to_sdk_mcp_tool(self, func_tool: FunctionTool) -> SdkMcpTool[Any]:
         """Convert a FunctionTool to an SDK MCP tool.
 
         Args:

python/packages/core/agent_framework/_agents.py

@@ -439,7 +439,7 @@ def as_tool(
         stream_callback: Callable[[AgentResponseUpdate], None]
         | Callable[[AgentResponseUpdate], Awaitable[None]]
         | None = None,
-    ) -> FunctionTool[BaseModel]:
+    ) -> FunctionTool:
         """Create a FunctionTool that wraps this agent.
 
         Keyword Args:
@@ -513,7 +513,7 @@ async def agent_wrapper(**kwargs: Any) -> str:
             # Create final text from accumulated updates
             return AgentResponse.from_updates(response_updates).text
 
-        agent_tool: FunctionTool[BaseModel] = FunctionTool(
+        agent_tool: FunctionTool = FunctionTool(
             name=tool_name,
             description=tool_description,
             func=agent_wrapper,
@@ -1258,17 +1258,12 @@ async def _log(level: types.LoggingLevel, data: Any) -> None:
         @server.list_tools()  # type: ignore
         async def _list_tools() -> list[types.Tool]:  # type: ignore
             """List all tools in the agent."""
-            # Get the JSON schema from the Pydantic model
-            schema = agent_tool.input_model.model_json_schema()
+            schema = agent_tool.parameters()
 
             tool = types.Tool(
                 name=agent_tool.name,
                 description=agent_tool.description,
-                inputSchema={
-                    "type": "object",
-                    "properties": schema.get("properties", {}),
-                    "required": schema.get("required", []),
-                },
+                inputSchema=schema,
             )
 
             await _log(level="debug", data=f"Agent tool: {agent_tool}")
@@ -1291,7 +1286,9 @@ async def _call_tool(  # type: ignore
 
             # Create an instance of the input model with the arguments
             try:
-                args_instance = agent_tool.input_model(**arguments)
+                args_instance: BaseModel | dict[str, Any] = (
+                    agent_tool.input_model(**arguments) if agent_tool.input_model is not None else arguments
+                )
                 result = await agent_tool.invoke(arguments=args_instance)
             except Exception as e:
                 raise McpError(

python/packages/core/agent_framework/_mcp.py

@@ -24,11 +24,9 @@
 from mcp.shared.context import RequestContext
 from mcp.shared.exceptions import McpError
 from mcp.shared.session import RequestResponder
-from pydantic import BaseModel, create_model
 
 from ._tools import (
     FunctionTool,
-    _build_pydantic_model_from_json_schema,
 )
 from ._types import (
     Content,
@@ -355,11 +353,14 @@ def _prepare_message_for_mcp(
     return messages
 
 
-def _get_input_model_from_mcp_prompt(prompt: types.Prompt) -> type[BaseModel]:
-    """Creates a Pydantic model from a prompt's parameters."""
+def _get_input_model_from_mcp_prompt(prompt: types.Prompt) -> dict[str, Any]:
+    """Get the input model from an MCP prompt.
+
+    Returns a JSON schema dictionary for prompt arguments.
+    """
     # Check if 'arguments' is missing or empty
     if not prompt.arguments:
-        return create_model(f"{prompt.name}_input")
+        return {"type": "object", "properties": {}}
 
     # Convert prompt arguments to JSON schema format
     properties: dict[str, Any] = {}
@@ -374,13 +375,10 @@ def _get_input_model_from_mcp_prompt(prompt: types.Prompt) -> type[BaseModel]:
         if prompt_argument.required:
             required.append(prompt_argument.name)
 
-    schema = {"properties": properties, "required": required}
-    return _build_pydantic_model_from_json_schema(prompt.name, schema)
-
-
-def _get_input_model_from_mcp_tool(tool: types.Tool) -> type[BaseModel]:
-    """Creates a Pydantic model from a tools parameters."""
-    return _build_pydantic_model_from_json_schema(tool.name, tool.inputSchema)
+    schema: dict[str, Any] = {"type": "object", "properties": properties}
+    if required:
+        schema["required"] = required
+    return schema
 
 
 def _normalize_mcp_name(name: str) -> str:
@@ -467,7 +465,7 @@ def __init__(
         self.session = session
         self.request_timeout = request_timeout
         self.client = client
-        self._functions: list[FunctionTool[Any]] = []
+        self._functions: list[FunctionTool] = []
         self.is_connected: bool = False
         self._tools_loaded: bool = False
         self._prompts_loaded: bool = False
@@ -476,7 +474,7 @@ def __str__(self) -> str:
         return f"MCPTool(name={self.name}, description={self.description})"
 
     @property
-    def functions(self) -> list[FunctionTool[Any]]:
+    def functions(self) -> list[FunctionTool]:
         """Get the list of functions that are allowed."""
         if not self.allowed_tools:
             return self._functions
@@ -744,7 +742,7 @@ async def load_prompts(self) -> None:
 
                 input_model = _get_input_model_from_mcp_prompt(prompt)
                 approval_mode = self._determine_approval_mode(local_name)
-                func: FunctionTool[BaseModel] = FunctionTool(
+                func: FunctionTool = FunctionTool(
                     func=partial(self.get_prompt, prompt.name),
                     name=local_name,
                     description=prompt.description or "",
@@ -785,15 +783,14 @@ async def load_tools(self) -> None:
                 if local_name in existing_names:
                     continue
 
-                input_model = _get_input_model_from_mcp_tool(tool)
                 approval_mode = self._determine_approval_mode(local_name)
                 # Create FunctionTools out of each tool
-                func: FunctionTool[BaseModel] = FunctionTool(
+                func: FunctionTool = FunctionTool(
                     func=partial(self.call_tool, tool.name),
                     name=local_name,
                     description=tool.description or "",
                     approval_mode=approval_mode,
-                    input_model=input_model,
+                    input_model=tool.inputSchema,
                 )
                 self._functions.append(func)
                 existing_names.add(local_name)

python/packages/core/agent_framework/_middleware.py

@@ -234,8 +234,8 @@ async def process(self, context: FunctionInvocationContext, call_next):
 
     def __init__(
         self,
-        function: FunctionTool[Any],
-        arguments: BaseModel,
+        function: FunctionTool,
+        arguments: BaseModel | Mapping[str, Any],
         metadata: Mapping[str, Any] | None = None,
         result: Any = None,
         kwargs: Mapping[str, Any] | None = None,