[BREAKING] Python: clean up kwargs across agents, chat clients, tools, and sessions (#4581)

* 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>

* clarified docstring

* feedback fixes

* Add unit tests for _docstrings.py build/apply helpers

Tests cover: no docstring source, no extra kwargs, appending to existing
Keyword Args section, inserting after Args, inserting in plain docstrings,
multiline descriptions, ordering, and apply_layered_docstring.

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

* Add test for propagate_session TypeError on non-AgentSession values

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

* Add tests for multi-content and empty UserInputRequiredException propagation

Cover the branching logic in _try_execute_function_calls for:
- Multiple user_input_request items in a single exception (extra_user_input_contents path)
- Empty contents list (fallback function_result path)

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

* Add tests for DurableAIAgent.get_session forwarding service_session_id

Verifies get_session correctly forwards service_session_id and session_id
to the executor's get_new_session, replacing the removed kwargs test.

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

* Simplify ag-ui test stub to read session from client_kwargs only

Remove dual-mode detection (client_kwargs vs raw kwargs fallback) from
the test mock. Session is now read exclusively from client_kwargs,
matching the settled public calling convention.

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

* updated create and get sessions in durable

* fixed docstrings

* fix test

* updated session handling

* updated from main

* updated tests

---------

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

python/packages/a2a/agent_framework_a2a/_agent.py

@@ -6,7 +6,7 @@ import base64
 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
@@ -226,6 +226,8 @@ class A2AAgent(AgentTelemetryLayer, BaseAgent):
         *,
         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,
@@ -238,17 +240,21 @@ class A2AAgent(AgentTelemetryLayer, BaseAgent):
         *,
         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,
@@ -261,17 +267,23 @@ class A2AAgent(AgentTelemetryLayer, BaseAgent):
         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"])