Python: [BREAKING] PR2 — Wire context provider pipeline, remove old types, update all consumers

python/packages/a2a/agent_framework_a2a/_agent.py

@@ -31,7 +31,7 @@
 from agent_framework import (
     AgentResponse,
     AgentResponseUpdate,
-    AgentThread,
+    AgentSession,
     BaseAgent,
     Content,
     ContinuationToken,
@@ -211,7 +211,7 @@ def run(
         messages: str | Message | Sequence[str | Message] | None = None,
         *,
         stream: Literal[False] = ...,
-        thread: AgentThread | None = None,
+        session: AgentSession | None = None,
         continuation_token: A2AContinuationToken | None = None,
         background: bool = False,
         **kwargs: Any,
@@ -223,7 +223,7 @@ def run(
         messages: str | Message | Sequence[str | Message] | None = None,
         *,
         stream: Literal[True],
-        thread: AgentThread | None = None,
+        session: AgentSession | None = None,
         continuation_token: A2AContinuationToken | None = None,
         background: bool = False,
         **kwargs: Any,
@@ -234,7 +234,7 @@ def run(
         messages: str | Message | Sequence[str | Message] | None = None,
         *,
         stream: bool = False,
-        thread: AgentThread | None = None,
+        session: AgentSession | None = None,
         continuation_token: A2AContinuationToken | None = None,
         background: bool = False,
         **kwargs: Any,
@@ -246,7 +246,7 @@ def run(
 
         Keyword Args:
             stream: Whether to stream the response. Defaults to False.
-            thread: The conversation thread associated with the message(s).
+            session: The conversation session associated with the message(s).
             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

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

@@ -18,20 +18,20 @@ def __init__(
         self,
         state_schema: Any | None = None,
         predict_state_config: dict[str, dict[str, str]] | None = None,
-        use_service_thread: bool = False,
+        use_service_session: bool = False,
         require_confirmation: bool = True,
     ):
         """Initialize agent configuration.
 
         Args:
             state_schema: Optional state schema for state management; accepts dict or Pydantic model/class
             predict_state_config: Configuration for predictive state updates
-            use_service_thread: Whether the agent thread is service-managed
+            use_service_session: Whether the agent session is service-managed
             require_confirmation: Whether predictive updates require user confirmation before applying
         """
         self.state_schema = self._normalize_state_schema(state_schema)
         self.predict_state_config = predict_state_config or {}
-        self.use_service_thread = use_service_thread
+        self.use_service_session = use_service_session
         self.require_confirmation = require_confirmation
 
     @staticmethod
@@ -77,7 +77,7 @@ def __init__(
         state_schema: Any | None = None,
         predict_state_config: dict[str, dict[str, str]] | None = None,
         require_confirmation: bool = True,
-        use_service_thread: bool = False,
+        use_service_session: bool = False,
     ):
         """Initialize the AG-UI compatible agent wrapper.
 
@@ -88,7 +88,7 @@ def __init__(
             state_schema: Optional state schema for state management; accepts dict or Pydantic model/class
             predict_state_config: Configuration for predictive state updates
             require_confirmation: Whether predictive updates require user confirmation before applying
-            use_service_thread: Whether the agent thread is service-managed
+            use_service_session: Whether the agent session is service-managed
         """
         self.agent = agent
         self.name = name or getattr(agent, "name", "agent")
@@ -97,7 +97,7 @@ def __init__(
         self.config = AgentConfig(
             state_schema=state_schema,
             predict_state_config=predict_state_config,
-            use_service_thread=use_service_thread,
+            use_service_session=use_service_session,
             require_confirmation=require_confirmation,
         )
 

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

@@ -171,11 +171,11 @@ class AGUIChatClient(
 
             client = AGUIChatClient(endpoint="http://localhost:8888/")
             agent = Agent(name="assistant", client=client)
-            thread = await agent.get_new_thread()
+            session = agent.create_session()
 
             # Agent automatically maintains history and sends full context
-            response = await agent.run("Hello!", thread=thread)
-            response2 = await agent.run("How are you?", thread=thread)
+            response = await agent.run("Hello!", session=session)
+            response2 = await agent.run("How are you?", session=session)
 
         Streaming usage:
 

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

@@ -27,7 +27,7 @@
     ToolCallStartEvent,
 )
 from agent_framework import (
-    AgentThread,
+    AgentSession,
     Content,
     Message,
     SupportsAgentRun,
@@ -809,12 +809,12 @@ async def run_agent_stream(
     register_additional_client_tools(agent, client_tools)
     tools = merge_tools(server_tools, client_tools)
 
-    # Create thread (with service thread support)
-    if config.use_service_thread:
+    # Create session (with service session support)
+    if config.use_service_session:
         supplied_thread_id = input_data.get("thread_id") or input_data.get("threadId")
-        thread = AgentThread(service_thread_id=supplied_thread_id)
+        session = AgentSession(service_session_id=supplied_thread_id)
     else:
-        thread = AgentThread()
+        session = AgentSession()
 
     # Inject metadata for AG-UI orchestration (Feature #2: Azure-safe truncation)
     base_metadata: dict[str, Any] = {
@@ -823,16 +823,16 @@ async def run_agent_stream(
     }
     if flow.current_state:
         base_metadata["current_state"] = flow.current_state
-    thread.metadata = _build_safe_metadata(base_metadata)  # type: ignore[attr-defined]
+    session.metadata = _build_safe_metadata(base_metadata)  # type: ignore[attr-defined]
 
     # Build run kwargs (Feature #6: Azure store flag when metadata present)
-    run_kwargs: dict[str, Any] = {"thread": thread}
+    run_kwargs: dict[str, Any] = {"session": session}
     if tools:
         run_kwargs["tools"] = tools
     # Filter out AG-UI internal metadata keys before passing to chat client
     # These are used internally for orchestration and should not be sent to the LLM provider
     client_metadata = {
-        k: v for k, v in (getattr(thread, "metadata", None) or {}).items() if k not in AG_UI_INTERNAL_METADATA_KEYS
+        k: v for k, v in (getattr(session, "metadata", None) or {}).items() if k not in AG_UI_INTERNAL_METADATA_KEYS
     }
     safe_metadata = _build_safe_metadata(client_metadata) if client_metadata else {}
     if safe_metadata:

python/packages/ag-ui/getting_started/client_with_agent.py

@@ -4,18 +4,18 @@
 
 This demonstrates the HYBRID pattern matching .NET AGUIClient implementation:
 
-1. AgentThread Pattern (like .NET):
-   - Create thread with agent.get_new_thread()
-   - Pass thread to agent.run(stream=True) on each turn
-   - Thread automatically maintains conversation history via message_store
+1. AgentSession Pattern (like .NET):
+   - Create session with agent.create_session()
+   - Pass session to agent.run(stream=True) on each turn
+   - Session maintains conversation context via context providers
 
 2. Hybrid Tool Execution:
    - AGUIChatClient uses function invocation mixin
    - Client-side tools (get_weather) can execute locally when server requests them
    - Server may also have its own tools that execute server-side
    - Both work together: server LLM decides which tool to call, decorator handles client execution
 
-This matches .NET pattern: thread maintains state, tools execute on appropriate side.
+This matches .NET pattern: session maintains state, tools execute on appropriate side.
 """
 
 from __future__ import annotations
@@ -59,13 +59,13 @@ async def main():
 
     This matches the .NET pattern from Program.cs where:
     - AIAgent agent = chatClient.CreateAIAgent(tools: [...])
-    - AgentThread thread = agent.GetNewThread()
-    - RunStreamingAsync(messages, thread)
+    - AgentSession session = agent.CreateSession()
+    - RunStreamingAsync(messages, session)
 
     Python equivalent:
     - agent = Agent(client=AGUIChatClient(...), tools=[...])
-    - thread = agent.get_new_thread()  # Creates thread with message_store
-    - agent.run(message, stream=True, thread=thread)  # Thread accumulates history
+    - session = agent.create_session()  # Creates session
+    - agent.run(message, stream=True, session=session)  # Session tracks context
     """
     server_url = os.environ.get("AGUI_SERVER_URL", "http://127.0.0.1:5100/")
 
@@ -74,7 +74,7 @@ async def main():
     print("=" * 70)
     print(f"\nServer: {server_url}")
     print("\nThis example demonstrates:")
-    print("  1. AgentThread maintains conversation state (like .NET)")
+    print("  1. AgentSession maintains conversation state (like .NET)")
     print("  2. Client-side tools execute locally via function invocation mixin")
     print("  3. Server may have additional tools that execute server-side")
     print("  4. HYBRID: Client and server tools work together simultaneously\n")
@@ -90,30 +90,30 @@ async def main():
                 tools=[get_weather],
             )
 
-            # Create a thread to maintain conversation state (like .NET AgentThread)
-            thread = agent.get_new_thread()
+            # Create a session to maintain conversation state (like .NET AgentSession)
+            session = agent.create_session()
 
             print("=" * 70)
             print("CONVERSATION WITH HISTORY")
             print("=" * 70)
 
             # Turn 1: Introduce
             print("\nUser: My name is Alice and I live in Seattle\n")
-            async for chunk in agent.run("My name is Alice and I live in Seattle", stream=True, thread=thread):
+            async for chunk in agent.run("My name is Alice and I live in Seattle", stream=True, session=session):
                 if chunk.text:
                     print(chunk.text, end="", flush=True)
             print("\n")
 
             # Turn 2: Ask about name (tests history)
             print("User: What's my name?\n")
-            async for chunk in agent.run("What's my name?", stream=True, thread=thread):
+            async for chunk in agent.run("What's my name?", stream=True, session=session):
                 if chunk.text:
                     print(chunk.text, end="", flush=True)
             print("\n")
 
             # Turn 3: Ask about location (tests history)
             print("User: Where do I live?\n")
-            async for chunk in agent.run("Where do I live?", stream=True, thread=thread):
+            async for chunk in agent.run("Where do I live?", stream=True, session=session):
                 if chunk.text:
                     print(chunk.text, end="", flush=True)
             print("\n")
@@ -123,64 +123,19 @@ async def main():
             async for chunk in agent.run(
                 "What's the weather forecast for today in Seattle?",
                 stream=True,
-                thread=thread,
+                session=session,
             ):
                 if chunk.text:
                     print(chunk.text, end="", flush=True)
             print("\n")
 
             # Turn 5: Test server-side tool (get_time_zone is server-side only)
             print("User: What time zone is Seattle in?\n")
-            async for chunk in agent.run("What time zone is Seattle in?", stream=True, thread=thread):
+            async for chunk in agent.run("What time zone is Seattle in?", stream=True, session=session):
                 if chunk.text:
                     print(chunk.text, end="", flush=True)
             print("\n")
 
-            # Show thread state
-            if thread.message_store:
-
-                def _preview_for_message(m) -> str:
-                    # Prefer plain text when present
-                    if getattr(m, "text", ""):
-                        t = m.text
-                        return (t[:60] + "...") if len(t) > 60 else t
-                    # Build from contents when no direct text
-                    parts: list[str] = []
-                    for c in getattr(m, "contents", []) or []:
-                        content_type = getattr(c, "type", None)
-                        if content_type == "function_call":
-                            args = getattr(c, "arguments", None)
-                            if isinstance(args, dict):
-                                try:
-                                    import json as _json
-
-                                    args_str = _json.dumps(args)
-                                except Exception:
-                                    args_str = str(args)
-                            else:
-                                args_str = str(args or "{}")
-                            parts.append(f"tool_call {getattr(c, 'name', '?')} {args_str}")
-                        elif content_type == "function_result":
-                            call_id = getattr(c, "call_id", "?")
-                            result = getattr(c, "result", None)
-                            parts.append(f"tool_result[{call_id}]: {str(result)[:40]}")
-                        elif content_type == "text":
-                            text = getattr(c, "text", None)
-                            if text:
-                                parts.append(text)
-                        else:
-                            typename = getattr(c, "type", c.__class__.__name__)
-                            parts.append(f"<{typename}>")
-                    preview = " | ".join(parts) if parts else ""
-                    return (preview[:60] + "...") if len(preview) > 60 else preview
-
-                messages = await thread.message_store.list_messages()
-                print(f"\n[THREAD STATE] {len(messages)} messages in thread's message_store")
-                for i, msg in enumerate(messages[-6:], 1):  # Show last 6
-                    role = msg.role if hasattr(msg.role, "value") else str(msg.role)
-                    text_preview = _preview_for_message(msg)
-                    print(f"  {i}. [{role}]: {text_preview}")
-
     except ConnectionError as e:
         print(f"\n\033[91mConnection Error: {e}\033[0m")
         print("\nMake sure an AG-UI server is running at the specified endpoint.")

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

@@ -11,7 +11,7 @@
 from agent_framework import (
     AgentResponse,
     AgentResponseUpdate,
-    AgentThread,
+    AgentSession,
     BaseChatClient,
     ChatOptions,
     ChatResponse,
@@ -49,8 +49,8 @@ def __init__(self, stream_fn: StreamFn, response_fn: ResponseFn | None = None) -
         super().__init__(function_middleware=[])
         self._stream_fn = stream_fn
         self._response_fn = response_fn
-        self.last_thread: AgentThread | None = None
-        self.last_service_thread_id: str | None = None
+        self.last_session: AgentSession | None = None
+        self.last_service_session_id: str | None = None
 
     @overload
     def get_response(
@@ -90,8 +90,8 @@ def get_response(
         options: OptionsCoT | ChatOptions[Any] | None = None,
         **kwargs: Any,
     ) -> Awaitable[ChatResponse[Any]] | ResponseStream[ChatResponseUpdate, ChatResponse[Any]]:
-        self.last_thread = kwargs.get("thread")
-        self.last_service_thread_id = self.last_thread.service_thread_id if self.last_thread else None
+        self.last_session = 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]],
             super().get_response(
@@ -178,7 +178,7 @@ def run(
         messages: str | Message | Sequence[str | Message] | None = None,
         *,
         stream: Literal[False] = ...,
-        thread: AgentThread | None = None,
+        session: AgentSession | None = None,
         **kwargs: Any,
     ) -> Awaitable[AgentResponse[Any]]: ...
 
@@ -188,7 +188,7 @@ def run(
         messages: str | Message | Sequence[str | Message] | None = None,
         *,
         stream: Literal[True],
-        thread: AgentThread | None = None,
+        session: AgentSession | None = None,
         **kwargs: Any,
     ) -> ResponseStream[AgentResponseUpdate, AgentResponse[Any]]: ...
 
@@ -197,7 +197,7 @@ def run(
         messages: str | Message | Sequence[str | Message] | None = None,
         *,
         stream: bool = False,
-        thread: AgentThread | None = None,
+        session: AgentSession | None = None,
         **kwargs: Any,
     ) -> Awaitable[AgentResponse[Any]] | ResponseStream[AgentResponseUpdate, AgentResponse[Any]]:
         if stream:
@@ -218,8 +218,8 @@ async def _get_response() -> AgentResponse[Any]:
 
         return _get_response()
 
-    def get_new_thread(self, **kwargs: Any) -> AgentThread:
-        return AgentThread()
+    def create_session(self, **kwargs: Any) -> AgentSession:
+        return AgentSession()
 
 
 # Fixtures