Python: [Bug]: usage_details not available in ContextProvider / HistoryProvider

[auroranockert]

Description

I would like to be able to look at the usage_details in the ContextProvider, but for some reason a fake AgentResponse is created right before calling the ContextProvider with limited information only to then right afterwards create the real AgentResponse with the full info.

agent-framework/python/packages/core/agent_framework/_agents.py

Lines 1032 to 1049 in 7f3a2ae

	await self._finalize_response(
	response=response,
	agent_name=context["agent_name"],
	session=context["session"],
	session_context=context["session_context"],
	suppress_response_id=context["suppress_response_id"],
	)
	return AgentResponse(
	messages=response.messages,
	response_id=None if context["suppress_response_id"] else response.response_id,
	created_at=response.created_at,
	usage_details=response.usage_details,
	value=response.value,
	response_format=context["chat_options"].get("response_format"),
	continuation_token=response.continuation_token,
	raw_representation=response,
	additional_properties=response.additional_properties,
	)

here the "real" AgentResponse is created, but inside self._finalize_response ) another AgentResponse is created right before.

agent-framework/python/packages/core/agent_framework/_agents.py

Lines 1411 to 1417 in 7f3a2ae

	session_context._response = AgentResponse( # type: ignore[assignment]
	messages=response.messages,
	response_id=None if suppress_response_id else response.response_id,
	)
	# Run after_run providers (reverse order)
	await self._run_after_providers(session=session, context=session_context)

If you just swap this around and make _parse_non_streaming_response look something like the following then it does what I assume is the correct behaviour?

    async def _parse_non_streaming_response(
        self,
        context: _RunContext,
        response: ChatResponse[Any],
    ) -> AgentResponse[Any]:
        """Finalize a non-streaming chat response into an AgentResponse."""
        if not response:
            raise AgentInvalidResponseException("Chat client did not return a response.")

        session = context["session"]
        session_context = context["session_context"]

        # Ensure that the author name is set for each message in the response.
        for message in response.messages:
            if message.author_name is None:
                message.author_name = context["agent_name"]

        # Propagate conversation_id back to session (e.g. thread ID from Assistants API).
        # For Responses-style APIs this can rotate every turn (response_id-based continuation),
        # so refresh when a newer value is returned.
        if (
            session
            and response.conversation_id
            and not is_local_history_conversation_id(response.conversation_id)
            and session.service_session_id != response.conversation_id
        ):
            session.service_session_id = response.conversation_id

        agent_response = AgentResponse(
            messages=response.messages,
            response_id=None if context["suppress_response_id"] else response.response_id,
            created_at=response.created_at,
            usage_details=response.usage_details,
            value=response.value,
            response_format=context["chat_options"].get("response_format"),
            continuation_token=response.continuation_token,
            raw_representation=response,
            additional_properties=response.additional_properties,
        )

        session_context._response = agent_response  # type: ignore[assignment]

        await self._run_after_providers(session=session, context=session_context)

        return agent_response

Code Sample

Error Messages / Stack Traces

Package Versions

agent-framework-core 1.19.0, but it is broken on main too

Python Version

Python 3.14.3

Additional Context

No response

[PratikWayase]

@eavanvalkenburg @rogerbarreto I'd like to work on this. Could you please assign it to me?

The proposed change is to move the full AgentResponse construction into _finalize_response and return it directly. The caller would simply return the result. This fixes both the streaming and non-streaming paths in one place and establishes a single source of truth for AgentResponse construction.

Before:

• _finalize_response → partial AgentResponse → after_providers → return
• Caller → full AgentResponse → return (too late for providers)

After:

• _finalize_response → full AgentResponse → after_providers → return
• Caller → returns the AgentResponse already built by _finalize_response

[eavanvalkenburg]

feel free to create a PR for this @PratikWayase I think the change makes sense

[PratikWayase]

Hi @eavanvalkenburg, @westey-m

I was able to reproduce the issue and put together a fix. The changes include:

• Inlining the response finalization logic into _parse_non_streaming_response so ContextProvider.after_run() receives the fully constructed AgentResponse.
• Updating the streaming path to populate the missing AgentResponse fields (created_at, usage_details, value, response_format, continuation_token, raw_representation, and additional_properties).
• Removing _finalize_response, since its responsibilities are now handled directly in the non-streaming path.
• Added regression tests for both the streaming and non-streaming paths. Test count increased from 145 to 147, and all tests are passing.

Would you mind taking a quick look at the approach? Does this seem like the right direction for a PR, or would you suggest any changes or a different implementation before I open one?

Thanks!

Changes

• python/packages/core/agent_framework/_agents.py

1. Update _parse_non_streaming_response

Removed the call to _finalize_response() and inlined its logic into _parse_non_streaming_response().

-        await self._finalize_response(
-            response=response,
-            agent_name=context["agent_name"],
-            session=context["session"],
-            session_context=context["session_context"],
-            suppress_response_id=context["suppress_response_id"],
-        )
-        return AgentResponse(
+        session = context["session"]
+        session_context = context["session_context"]
+
+        # Ensure that the author name is set for each message in the response.
+        for message in response.messages:
+            if message.author_name is None:
+                message.author_name = context["agent_name"]
+
+        # Propagate conversation_id back to session.
+        if (
+            session
+            and response.conversation_id
+            and not is_local_history_conversation_id(response.conversation_id)
+            and session.service_session_id != response.conversation_id
+        ):
+            session.service_session_id = response.conversation_id
+
+        agent_response = AgentResponse(
             messages=response.messages,
             response_id=None if context["suppress_response_id"] else response.response_id,
             created_at=response.created_at,
             usage_details=response.usage_details,
             value=response.value,
             response_format=context["chat_options"].get("response_format"),
             continuation_token=response.continuation_token,
             raw_representation=response,
             additional_properties=response.additional_properties,
         )
+
+        session_context._response = agent_response
+        await self._run_after_providers(
+            session=session,
+            context=session_context,
+        )
+
+        return agent_response

This ensures that ContextProvider.after_run() receives the exact same fully populated AgentResponse object that is returned to the caller.

---

2. Update _parse_streaming_response

Populate the remaining AgentResponse fields before invoking after_providers.

session_context._response = AgentResponse(
    messages=response.messages,
    response_id=None if suppress_response_id else response.response_id,
+   created_at=response.created_at,
+   usage_details=response.usage_details,
+   value=response.value,
+   response_format=context["chat_options"].get("response_format"),
+   continuation_token=response.continuation_token,
+   raw_representation=response,
+   additional_properties=response.additional_properties,
)

This makes the streaming and non-streaming code paths consistent.

---

3. Remove _finalize_response

Since its responsibilities are now handled directly inside _parse_non_streaming_response(), _finalize_response() is no longer needed and has been removed.

---

4. Add Regression Tests

• python/packages/core/tests/core/test_agents.py

Added two regression tests to ensure that ContextProvider.after_run() always receives the same fully populated AgentResponse object that is returned to the caller.

Non-streaming

async def test_context_provider_receives_full_agent_response_non_streaming():
    """
    ContextProviders should see the full AgentResponse (with usage_details,
    raw_representation, etc.), not a minimal stub. The object seen by the
    ContextProvider must be the exact same object returned to the caller.
    """
    captured_response = None

    class CapturingContextProvider(ContextProvider):
        def __init__(self):
            super().__init__(source_id="test_capture")

        async def after_run(self, *, agent, session, context, state) -> None:
            nonlocal captured_response
            captured_response = context.response

    client = MockBaseChatClient()
    agent = Agent(
        name="test_agent",
        client=client,
        context_providers=[CapturingContextProvider()],
    )

    response = await agent.run(
        messages=[Message(role="user", contents="hi")],
        stream=False,
    )

    assert captured_response is not None
    assert captured_response is response
    assert captured_response.raw_representation is not None

Streaming

async def test_context_provider_receives_full_agent_response_streaming():
    """Regression test for the streaming path."""
    captured_response = None

    class CapturingContextProvider(ContextProvider):
        def __init__(self):
            super().__init__(source_id="test_capture")

        async def after_run(self, *, agent, session, context, state) -> None:
            nonlocal captured_response
            captured_response = context.response

    client = MockBaseChatClient()
    agent = Agent(
        name="test_agent",
        client=client,
        context_providers=[CapturingContextProvider()],
    )

    stream = agent.run(
        messages=[Message(role="user", contents="hi")],
        stream=True,
    )

    async for _ in stream:
        pass

    assert captured_response is not None
    assert captured_response.raw_representation is not None

These regression tests verify that:

• ContextProvider.after_run() receives a fully populated AgentResponse.
• The non-streaming path exposes the exact same AgentResponse instance returned by agent.run().
• The streaming path also exposes a fully initialized AgentResponse, including raw_representation, preventing regressions where context providers observe only a partial response.