Python: clean up kwargs across agents, chat clients, tools, and sessions (#3642)

Audit and refactor public **kwargs usage across core agents, chat clients,
tools, sessions, and provider packages per the migration strategy codified
in CODING_STANDARD.md.

Key changes:
- Add explicit runtime buckets: function_invocation_kwargs and client_kwargs
  on RawAgent.run() and chat client get_response() layers.
- Refactor FunctionTool to prefer explicit ctx: FunctionInvocationContext
  injection; legacy **kwargs tools still work via _forward_runtime_kwargs.
- Refactor Agent.as_tool() to use direct JSON schema, always-streaming
  wrapper, approval_mode parameter, and UserInputRequiredException
  propagation (integrates PR #4568 behavior).
- Remove implicit session bleeding into FunctionInvocationContext; tools
  that need a session must receive it via function_invocation_kwargs.
- Lower chat-client layers after FunctionInvocationLayer accept only
  compatibility **kwargs (client_kwargs flattened, function_invocation_kwargs
  ignored).
- Add layered docstring composition from Raw... implementations via
  _docstrings.py helper.
- Clean up provider constructors to use explicit additional_properties.
- Deprecation warnings on legacy direct kwargs paths.
- Update samples, tests, and typing across all 23 packages.

Resolves #3642

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: eavanvalkenburg

docs/decisions/0001-agent-run-response.md

@@ -64,7 +64,7 @@ Approaches observed from the compared SDKs:
 | AutoGen | **Approach 1** Separates messages into Agent-Agent (maps to Primary) and Internal (maps to Secondary) and these are returned as separate properties on the agent response object.  See [types of messages](https://microsoft.github.io/autogen/stable/user-guide/agentchat-user-guide/tutorial/messages.html#types-of-messages) and [Response](https://microsoft.github.io/autogen/stable/reference/python/autogen_agentchat.base.html#autogen_agentchat.base.Response) | **Approach 2** Returns a stream of internal events and the last item is a Response object. See [ChatAgent.on_messages_stream](https://microsoft.github.io/autogen/stable/reference/python/autogen_agentchat.base.html#autogen_agentchat.base.ChatAgent.on_messages_stream) |
 | OpenAI Agent SDK | **Approach 1** Separates new_items (Primary+Secondary) from final output (Primary) as separate properties on the [RunResult](https://github.com/openai/openai-agents-python/blob/main/src/agents/result.py#L39) | **Approach 1** Similar to non-streaming, has a way of streaming updates via a method on the response object which includes all data, and then a separate final output property on the response object which is populated only when the run is complete. See [RunResultStreaming](https://github.com/openai/openai-agents-python/blob/main/src/agents/result.py#L136) |
 | Google ADK | **Approach 2** [Emits events](https://google.github.io/adk-docs/runtime/#step-by-step-breakdown) with [FinalResponse](https://github.com/google/adk-java/blob/main/core/src/main/java/com/google/adk/events/Event.java#L232) true (Primary) / false (Secondary) and callers have to filter out those with false to get just the final response message | **Approach 2** Similar to non-streaming except [events](https://google.github.io/adk-docs/runtime/#streaming-vs-non-streaming-output-partialtrue) are emitted with [Partial](https://github.com/google/adk-java/blob/main/core/src/main/java/com/google/adk/events/Event.java#L133) true to indicate that they are streaming messages. A final non partial event is also emitted. |
-| AWS (Strands) | **Approach 3** Returns an [AgentResult](https://strandsagents.com/latest/documentation/docs/api-reference/python/agent/agent_result/) (Primary) with messages and a reason for the run's completion. | **Approach 2** [Streams events](https://strandsagents.com/latest/documentation/docs/api-reference/python/agent/agent/#strands.agent.agent.Agent.stream_async) (Primary+Secondary) including, response text, current_tool_use, even data from "callbacks" (strands plugins) |
+| AWS (Strands) | **Approach 3** Returns an [AgentResult](https://strandsagents.com/docs/api/python/strands.agent.agent_result/) (Primary) with messages and a reason for the run's completion. | **Approach 2** [Streams events](https://strandsagents.com/docs/api/python/strands.agent.agent/) (Primary+Secondary) including, response text, current_tool_use, even data from "callbacks" (strands plugins) |
 | LangGraph | **Approach 2** A mixed list of all [messages](https://langchain-ai.github.io/langgraph/agents/run_agents/#output-format) | **Approach 2** A mixed list of all [messages](https://langchain-ai.github.io/langgraph/agents/run_agents/#output-format) |
 | Agno | **Combination of various approaches** Returns a [RunResponse](https://docs.agno.com/reference/agents/run-response) object with text content, messages (essentially chat history including inputs and instructions), reasoning and thinking text properties. Secondary events could potentially be extracted from messages. | **Approach 2** Returns [RunResponseEvent](https://docs.agno.com/reference/agents/run-response#runresponseevent-types-and-attributes) objects including tool call, memory update, etc, information, where the [RunResponseCompletedEvent](https://docs.agno.com/reference/agents/run-response#runresponsecompletedevent) has similar properties to RunResponse|
 | A2A | **Approach 3** Returns a [Task or Message](https://a2aproject.github.io/A2A/latest/specification/#71-messagesend) where the message is the final result (Primary) and task is a reference to a long running process. | **Approach 2** Returns a [stream](https://a2aproject.github.io/A2A/latest/specification/#72-messagestream) that contains task updates (Secondary) and a final message (Primary) |
@@ -496,7 +496,7 @@ We need to decide what AIContent types, each agent response type will be mapped
 |-|-|
 | AutoGen | **Approach 1** Supports [configuring an agent](https://microsoft.github.io/autogen/stable/user-guide/agentchat-user-guide/tutorial/agents.html#structured-output) at agent creation. |
 | Google ADK | **Approach 1** Both [input and output schemas can be specified for LLM Agents](https://google.github.io/adk-docs/agents/llm-agents/#structuring-data-input_schema-output_schema-output_key) at construction time. This option is specific to this agent type and other agent types do not necessarily support |
-| AWS (Strands) | **Approach 2** Supports a special invocation method called [structured_output](https://strandsagents.com/latest/documentation/docs/api-reference/python/agent/agent/#strands.agent.agent.Agent.structured_output) |
+| AWS (Strands) | **Approach 2** Supports a special invocation method called [structured_output](https://strandsagents.com/docs/api/python/strands.agent.agent/) |
 | LangGraph | **Approach 1** Supports [configuring an agent](https://langchain-ai.github.io/langgraph/agents/agents/?h=structured#6-configure-structured-output) at agent construction time, and a [structured response](https://langchain-ai.github.io/langgraph/agents/run_agents/#output-format) can be retrieved as a special property on the agent response |
 | Agno | **Approach 1** Supports [configuring an agent](https://docs.agno.com/input-output/structured-output/agent) at agent construction time |
 | A2A | **Informal Approach 2** Doesn't formally support schema negotiation, but [hints can be provided via metadata](https://a2a-protocol.org/latest/specification/#97-structured-data-exchange-requesting-and-providing-json) at invocation time |
@@ -508,7 +508,7 @@ We need to decide what AIContent types, each agent response type will be mapped
 |-|-|
 | AutoGen | Supports a [stop reason](https://microsoft.github.io/autogen/stable/reference/python/autogen_agentchat.base.html#autogen_agentchat.base.TaskResult.stop_reason) which is a freeform text string |
 | Google ADK | [No equivalent present](https://github.com/google/adk-python/blob/main/src/google/adk/events/event.py) |
-| AWS (Strands) | Exposes a [stop_reason](https://strandsagents.com/latest/documentation/docs/api-reference/python/types/event_loop/#strands.types.event_loop.StopReason) property on the [AgentResult](https://strandsagents.com/latest/documentation/docs/api-reference/python/agent/agent_result/) class with options that are tied closely to LLM operations. |
+| AWS (Strands) | Exposes a [stop_reason](https://strandsagents.com/docs/api/python/strands.types.event_loop/) property on the [AgentResult](https://strandsagents.com/docs/api/python/strands.agent.agent_result/) class with options that are tied closely to LLM operations. |
 | LangGraph | No equivalent present, output contains only [messages](https://langchain-ai.github.io/langgraph/agents/run_agents/#output-format) |
 | Agno | [No equivalent present](https://docs.agno.com/reference/agents/run-response) |
 | A2A | No equivalent present, response only contains a [message](https://a2a-protocol.org/latest/specification/#64-message-object) or [task](https://a2a-protocol.org/latest/specification/#61-task-object). |

python/CODING_STANDARD.md

@@ -127,7 +127,12 @@ def create_agent(name: str, tool_mode: Literal['auto', 'required', 'none'] | Cha
 Avoid `**kwargs` unless absolutely necessary. It should only be used as an escape route, not for well-known flows of data:
 
 - **Prefer named parameters**: If there are known extra arguments being passed, use explicit named parameters instead of kwargs
+- **Prefer purpose-specific buckets over generic kwargs**: If a flexible payload is still needed, use an explicit named parameter such as `additional_properties`, `function_invocation_kwargs`, or `client_kwargs` rather than a blanket `**kwargs`
 - **Subclassing support**: kwargs is acceptable in methods that are part of classes designed for subclassing, allowing subclass-defined kwargs to pass through without issues. In this case, clearly document that kwargs exists for subclass extensibility and not for passing arbitrary data
+- **Make known flows explicit first**: For abstract hooks, move known data flows into explicit parameters before leaving `**kwargs` behind for subclass extensibility (for example, prefer `state=` explicitly instead of passing it through kwargs)
+- **Prefer explicit metadata containers**: For constructors that expose metadata, prefer an explicit `additional_properties` parameter.
+- **Keep SDK passthroughs narrow and documented**: A kwargs escape hatch may be acceptable for provider helper APIs that pass through to a large or unstable external SDK surface, but it should be documented as SDK passthrough and revisited regularly
+- **Do not keep passthrough kwargs on wrappers that do not use them**: Convenience wrappers and session helpers should not accept generic kwargs merely to forward or ignore them
 - **Remove when possible**: In other cases, removing kwargs is likely better than keeping it
 - **Separate kwargs by purpose**: When combining kwargs for multiple purposes, use specific parameters like `client_kwargs: dict[str, Any]` instead of mixing everything in `**kwargs`
 - **Always document**: If kwargs must be used, always document how it's used, either by referencing external documentation or explaining its purpose

python/packages/a2a/agent_framework_a2a/_agent.py

@@ -6,7 +6,7 @@
 import json
 import re
 import uuid
-from collections.abc import AsyncIterable, Awaitable, Sequence
+from collections.abc import AsyncIterable, Awaitable, Mapping, Sequence
 from typing import Any, Final, Literal, TypeAlias, overload
 
 import httpx
@@ -218,6 +218,8 @@ def run(
         *,
         stream: Literal[False] = ...,
         session: AgentSession | None = None,
+        function_invocation_kwargs: Mapping[str, Any] | None = None,
+        client_kwargs: Mapping[str, Any] | None = None,
         continuation_token: A2AContinuationToken | None = None,
         background: bool = False,
         **kwargs: Any,
@@ -230,17 +232,21 @@ def run(
         *,
         stream: Literal[True],
         session: AgentSession | None = None,
+        function_invocation_kwargs: Mapping[str, Any] | None = None,
+        client_kwargs: Mapping[str, Any] | None = None,
         continuation_token: A2AContinuationToken | None = None,
         background: bool = False,
         **kwargs: Any,
     ) -> ResponseStream[AgentResponseUpdate, AgentResponse[Any]]: ...
 
-    def run(
+    def run(  # pyright: ignore[reportIncompatibleMethodOverride]
         self,
         messages: AgentRunInputs | None = None,
         *,
         stream: bool = False,
         session: AgentSession | None = None,
+        function_invocation_kwargs: Mapping[str, Any] | None = None,
+        client_kwargs: Mapping[str, Any] | None = None,
         continuation_token: A2AContinuationToken | None = None,
         background: bool = False,
         **kwargs: Any,
@@ -253,17 +259,23 @@ def run(
         Keyword Args:
             stream: Whether to stream the response. Defaults to False.
             session: The conversation session associated with the message(s).
+            function_invocation_kwargs: Present for compatibility with the shared agent interface.
+                A2AAgent does not use these values directly.
+            client_kwargs: Present for compatibility with the shared agent interface.
+                A2AAgent does not use these values directly.
+            kwargs: Additional compatibility keyword arguments.
+                A2AAgent does not use these values directly.
             continuation_token: Optional token to resume a long-running task
                 instead of starting a new one.
             background: When True, in-progress task updates surface continuation
                 tokens so the caller can poll or resubscribe later. When False
                 (default), the agent internally waits for the task to complete.
-            kwargs: Additional keyword arguments.
 
         Returns:
             When stream=False: An Awaitable[AgentResponse].
             When stream=True: A ResponseStream of AgentResponseUpdate items.
         """
+        del function_invocation_kwargs, client_kwargs, kwargs
         if continuation_token is not None:
             a2a_stream: AsyncIterable[A2AStreamItem] = self.client.resubscribe(
                 TaskIdParams(id=continuation_token["task_id"])

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

@@ -220,7 +220,6 @@ def __init__(
         additional_properties: dict[str, Any] | None = None,
         middleware: Sequence[ChatAndFunctionMiddlewareTypes] | None = None,
         function_invocation_configuration: FunctionInvocationConfiguration | None = None,
-        **kwargs: Any,
     ) -> None:
         """Initialize the AG-UI chat client.
 
@@ -231,13 +230,11 @@ def __init__(
             additional_properties: Additional properties to store
             middleware: Optional middleware to apply to the client.
             function_invocation_configuration: Optional function invocation configuration override.
-            **kwargs: Additional arguments passed to BaseChatClient
         """
         super().__init__(
             additional_properties=additional_properties,
             middleware=middleware,
             function_invocation_configuration=function_invocation_configuration,
-            **kwargs,
         )
         self._http_service = AGUIHttpService(
             endpoint=endpoint,

python/packages/ag-ui/tests/ag_ui/conftest.py

@@ -98,7 +98,11 @@ def get_response(
         options: OptionsCoT | ChatOptions[Any] | None = None,
         **kwargs: Any,
     ) -> Awaitable[ChatResponse[Any]] | ResponseStream[ChatResponseUpdate, ChatResponse[Any]]:
-        self.last_session = kwargs.get("session")
+        compatibility_client_kwargs = kwargs.get("client_kwargs")
+        if isinstance(compatibility_client_kwargs, Mapping):
+            self.last_session = cast(AgentSession | None, compatibility_client_kwargs.get("session"))
+        else:
+            self.last_session = cast(AgentSession | None, kwargs.get("session"))
         self.last_service_session_id = self.last_session.service_session_id if self.last_session else None
         return cast(
             Awaitable[ChatResponse[Any]] | ResponseStream[ChatResponseUpdate, ChatResponse[Any]],

python/packages/ag-ui/tests/ag_ui/test_agent_wrapper_comprehensive.py

@@ -702,14 +702,9 @@ async def test_agent_with_use_service_session_is_true(streaming_chat_client_stub
     """Test that when use_service_session is True, the AgentSession used to run the agent is set to the service session ID."""
     from agent_framework.ag_ui import AgentFrameworkAgent
 
-    request_service_session_id: str | None = None
-
     async def stream_fn(
         messages: MutableSequence[Message], chat_options: ChatOptions, **kwargs: Any
     ) -> AsyncIterator[ChatResponseUpdate]:
-        nonlocal request_service_session_id
-        session = kwargs.get("session")
-        request_service_session_id = session.service_session_id if session else None
         yield ChatResponseUpdate(
             contents=[Content.from_text(text="Response")], response_id="resp_67890", conversation_id="conv_12345"
         )
@@ -719,11 +714,22 @@ async def stream_fn(
 
     input_data = {"messages": [{"role": "user", "content": "Hi"}], "thread_id": "conv_123456"}
 
+    # Spy on agent.run to capture the session kwarg at call time (before streaming mutates it)
+    captured_service_session_id: str | None = None
+    original_run = agent.run
+
+    def capturing_run(*args: Any, **kwargs: Any) -> Any:
+        nonlocal captured_service_session_id
+        session = kwargs.get("session")
+        captured_service_session_id = session.service_session_id if session else None
+        return original_run(*args, **kwargs)
+
+    agent.run = capturing_run  # type: ignore[assignment, method-assign]
+
     events: list[Any] = []
     async for event in wrapper.run(input_data):
         events.append(event)
-    request_service_session_id = agent.client.last_service_session_id
-    assert request_service_session_id == "conv_123456"  # type: ignore[attr-defined] (service_session_id should be set)
+    assert captured_service_session_id == "conv_123456"
 
 
 async def test_function_approval_mode_executes_tool(streaming_chat_client_stub):

python/packages/anthropic/agent_framework_anthropic/_chat_client.py

@@ -228,11 +228,11 @@ def __init__(
         model_id: str | None = None,
         anthropic_client: AsyncAnthropic | None = None,
         additional_beta_flags: list[str] | None = None,
+        additional_properties: dict[str, Any] | None = None,
         middleware: Sequence[ChatAndFunctionMiddlewareTypes] | None = None,
         function_invocation_configuration: FunctionInvocationConfiguration | None = None,
         env_file_path: str | None = None,
         env_file_encoding: str | None = None,
-        **kwargs: Any,
     ) -> None:
         """Initialize an Anthropic Agent client.
 
@@ -244,11 +244,11 @@ def __init__(
                 For instance if you need to set a different base_url for testing or private deployments.
             additional_beta_flags: Additional beta flags to enable on the client.
                 Default flags are: "mcp-client-2025-04-04", "code-execution-2025-08-25".
+            additional_properties: Additional properties stored on the client instance.
             middleware: Optional middleware to apply to the client.
             function_invocation_configuration: Optional function invocation configuration override.
             env_file_path: Path to environment file for loading settings.
             env_file_encoding: Encoding of the environment file.
-            kwargs: Additional keyword arguments passed to the parent class.
 
         Examples:
             .. code-block:: python
@@ -319,9 +319,9 @@ class MyOptions(AnthropicChatOptions, total=False):
 
         # Initialize parent
         super().__init__(
+            additional_properties=additional_properties,
             middleware=middleware,
             function_invocation_configuration=function_invocation_configuration,
-            **kwargs,
         )
 
         # Initialize instance variables

python/packages/azure-ai-search/tests/test_aisearch_context_provider.py

@@ -16,6 +16,18 @@
 # -- Helpers -------------------------------------------------------------------
 
 
+@pytest.fixture(autouse=True)
+def clear_azure_search_env(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Keep tests isolated from ambient Azure Search environment variables."""
+    for key in (
+        "AZURE_SEARCH_ENDPOINT",
+        "AZURE_SEARCH_INDEX_NAME",
+        "AZURE_SEARCH_KNOWLEDGE_BASE_NAME",
+        "AZURE_SEARCH_API_KEY",
+    ):
+        monkeypatch.delenv(key, raising=False)
+
+
 class MockSearchResults:
     """Async-iterable mock for Azure SearchClient.search() results."""