Python: Add hosting protocol helper surface

[eavanvalkenburg]

Motivation & Context

The Python hosting packages are moving toward a lighter helper-first shape where Agent Framework owns protocol-to-run conversion and shared execution state, while apps keep their native web framework routes, auth, middleware, background work, deployment shape, and response construction.

This implements the first slice of that direction for the existing hosting and hosting-responses packages, removes the earlier unreleased host/channel surface, and updates the local Responses samples so reviewers can inspect the intended developer experience for both agents and workflows.

Description & Review Guide

• What are the major changes?
  • Adds AgentState, WorkflowState, SessionStore, AgentRunArgs, and WorkflowRunArgs to agent-framework-hosting.
  • Adds Responses helpers: create_response_id, responses_session_id, responses_to_run, responses_from_run, and responses_from_streaming_run.
  • Moves full Responses output rendering into the helper path so rich AF content maps to OpenAI Responses output objects.
  • Removes the unreleased host/channel implementation and the old hosting-telegram package from the Python workspace.
  • Updates local_responses to use native FastAPI routing, AgentState, SessionStore, non-streaming JSON, and streaming via ResponseStream.
  • Rebuilds local_responses_workflow as a helper-first FastAPI workflow sample using WorkflowState, explicit FileCheckpointStorage, and an app-owned checkpoint cursor.
• What is the impact of these changes?
  • Introduces the helper/state surface as the package contract and narrows the hosting package away from route aggregation, channel registration, host-owned auth, delivery, and command handling.
  • Makes the Responses samples available for feedback on the new app-owned hosting shape before additional protocol packages are rebuilt on the same seam.
• What do you want reviewers to focus on?
  • The naming and ergonomics of AgentState, WorkflowState, SessionStore, and the Responses helper functions.
  • Whether the full Responses rendering logic belongs in the helper module as implemented here.
  • Whether the samples make the developer-owned responsibilities clear enough, especially authN/authZ and session/checkpoint continuation.

Related Issue

Closes #6585. Refs #6588 for the follow-up Telegram helper-first work. Related ADR: #6837. Checked for similar open PRs; existing hosting/channel PRs target the previous channel model or other protocols.

Contribution Checklist

• The code builds clean without any errors or warnings
• All unit tests pass, and I have added new tests where possible
• The PR follows the Contribution Guidelines
• This PR is linked to an issue and there is no other open PR for this issue (see Related Issue above).
• This is not a breaking change. If it is a breaking change, add the breaking change label (or add "[BREAKING]" to the title prefix, before or after any language prefix) — a workflow keeps the label and title prefix in sync automatically.

[github-actions]

Python Test Coverage Report •

File	Stmts	Miss	Cover	Missing
packages/hosting-responses/agent_framework_hosting_responses
_parsing.py	473	206	56%	96, 119, 231, 235, 250–253, 255–260, 262–265, 269–279, 288–292, 323–325, 327–331, 335–338, 340–343, 345–348, 362–393, 457, 471, 491, 494–497, 499, 518–525, 527, 543–545, 559, 571, 583, 598–605, 607, 618–619, 633, 642–644, 654, 659, 668, 672, 676–677, 682–685, 692, 696–704, 708–713, 717–724, 728–732, 736–738, 742–744, 748–750, 755, 757, 762–768, 774–777, 785–789, 795, 797–799, 804–810, 814–817, 894
packages/hosting/agent_framework_hosting
_state.py	119	13	89%	200, 213–215, 232, 293, 321–324, 336–338
TOTAL	43952	5262	88%

Python Unit Test Overview

Tests	Skipped	Failures	Errors	Time
8786	33 💤	0 ❌	0 🔥	2m 19s ⏱️

[Copilot]

Pull request overview

This PR introduces a helper-first Python hosting surface: agent-framework-hosting now provides shared execution/session state primitives for app-owned routes, and agent-framework-hosting-responses provides OpenAI Responses protocol conversion helpers (request → run args, run result → Responses JSON/SSE). It also updates the local Responses hosting sample to use native FastAPI routing to demonstrate the intended developer experience.

Changes:

• Added AgentFrameworkState, SessionStore, and run-args TypedDicts (AgentRunArgs, WorkflowRunArgs) to agent-framework-hosting.
• Added Responses helper functions (responses_to_run, responses_from_run, responses_stream_events_from_run, responses_session_id, create_response_id) and expanded output rendering to map richer AF content into Responses output items.
• Updated the local_responses sample + docs to use FastAPI routes and the new helper/state surface (streaming via SSE and non-streaming JSON).

Reviewed changes

Copilot reviewed 13 out of 13 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
python/samples/04-hosting/af-hosting/README.md	Updates sample index text to reflect the helper-first + FastAPI route approach.
python/samples/04-hosting/af-hosting/local_responses/README.md	Reframes the sample around Responses helpers + app-owned FastAPI routes instead of channel run hooks.
python/samples/04-hosting/af-hosting/local_responses/pyproject.toml	Adds FastAPI dependency for the updated sample.
python/samples/04-hosting/af-hosting/local_responses/call_server_af.py	Updates the AF-backed client script to match the new (non-streaming) interaction pattern.
python/samples/04-hosting/af-hosting/local_responses/app.py	Replaces channel host usage with a native FastAPI /responses route using AgentFrameworkState + Responses helpers.
python/packages/hosting/tests/hosting/test_state.py	Adds unit tests for SessionStore and AgentFrameworkState behaviors.
python/packages/hosting/README.md	Updates package positioning and quickstart toward app-owned hosting with shared state helpers.
python/packages/hosting/agent_framework_hosting/_state.py	Introduces the new state/session primitives and TypedDict run-args surfaces.
python/packages/hosting/agent_framework_hosting/init.py	Exports the new state/session/run-args surface from the package.
python/packages/hosting-responses/tests/hosting_responses/test_parsing.py	Adds tests for new Responses helper functions and richer output-item mapping.
python/packages/hosting-responses/README.md	Updates docs from “channel” framing to “helpers for app-owned routes” with an example.
python/packages/hosting-responses/agent_framework_hosting_responses/_parsing.py	Adds helper APIs and full Responses output/SSE rendering backed by OpenAI SDK Pydantic models.
python/packages/hosting-responses/agent_framework_hosting_responses/init.py	Re-exports the new helper functions.

[github-actions]

Automated Code Review

Reviewers: 5 | Confidence: 85%

✓ Correctness

The PR adds hosting protocol helpers (AgentFrameworkState, SessionStore, and Responses conversion functions) with a clean separation between framework-owned state and app-owned routing. The implementation is correct: ResponseStream iteration semantics are properly handled (auto-finalization on consumption, safe re-call of get_final_response), AgentResponse.messages is always a list[Message] matching the Sequence check in result_to_output_items, and the session_id routing logic (conv prefix → conversation field, resp_ prefix → omitted) is consistent. No bugs, race conditions, or incorrect API usage found.

✓ Security Reliability

The PR introduces hosting protocol helpers and a shared state surface. The code is generally well-structured with appropriate input validation (empty session_id checks, type guards on body fields). One moderate reliability concern: AgentFrameworkState.get_target() has a race condition for async target factories under concurrent access — the check-then-await-then-set sequence lacks synchronization, so concurrent calers can bypass the intended once-only initialization promise.

✓ Test Coverage

Test coverage for the new helper functions is reasonable for the core happy paths (text, reasoning, function_call, function_result, streaming) but leaves a substantial rendering surface untested. The ~600 lines of content-type rendering code in _parsing.py (code_interpreter, image_generation, mcp, shell, approval, media) are duplicated from _channel.py and NOT covered by the existing channel tests since those exercise the channel's own copy. The AgentFrameworkState tests cover the main paths but miss reset_session() and documented error paths (ValueError, TypeError).

✓ Failure Modes

The new helper surface is well-structured and the rendering logic is thorough. The primary failure-mode concern is a TOCTOU race in AgentFrameworkState.get_target() and get_session_store() when async target factories are used under concurrent requests: the await at line 145 of _state.py creates a yield point between the cache check and cache write, which can lead to duplicate target creation and, more critically, split SessionStore instances where sessions created in the first store are silently lost to subsequent requests that use the second (cached) store.

✗ Design Approach

I found two design-level gaps in the new helper-first surface. First, the sample and README wire SessionStore directly to previous_response_id, which only preserves continuity for the second turn and silently starts a fresh AgentSession on turn 3+ unless callers also invent a stable conversation_id. Second, the new streaming helper only emits text deltas, so streamed reasoning/tool-call updates are dropped until the final response.completed event even though the existing channel/client in this repo already support richer Responses event shapes.

Flagged Issues

• Session continuity in the helper-first sample breaks after turn 2 because local_responses/app.py:111 keys SessionStore by responses_session_id(...), and _parsing.py:177-196 returns the changing previous_response_id while _state.py:31-47 only reuses sessions for identical keys.
• The new SSE helper in _parsing.py:906-914 streams only response.output_text.delta, while the existing Responses channel already emits per-content events in _channel.py:368-456 and 447-756; streamed tool/reasoning updates are therefore silently suppressed until completion.

---

Automated review by eavanvalkenburg's agents

[github-actions]

Flagged issue

Session continuity in the helper-first sample breaks after turn 2 because local_responses/app.py:111 keys SessionStore by responses_session_id(...), and _parsing.py:177-196 returns the changing previous_response_id while _state.py:31-47 only reuses sessions for identical keys.

---

Source: automated DevFlow PR review

[github-actions]

Flagged issue

The new SSE helper in _parsing.py:906-914 streams only response.output_text.delta, while the existing Responses channel already emits per-content events in _channel.py:368-456 and 447-756; streamed tool/reasoning updates are therefore silently suppressed until completion.

---

Source: automated DevFlow PR review

[github-actions]

Automated Code Review

Reviewers: 5 | Confidence: 77%

✓ Correctness

The code is consistent, error handling is present, and the samples clearly document their non-production status. The workflow sample's checkpoint cursor logic has a minor fragility around the conversation_id prefix heuristic, but it works correctly for the documented and demonstrated use cases. No blocking correctness issues found.

✓ Security Reliability

The new _state.py introduces AgentState and WorkflowState classes with a get_target() method that has a race condition when target is a raw awaitable (coroutine object): concurrent callers before the first resolution completes will both attempt to await the same coroutine, causing a RuntimeError. The SessionStore and session management have proper input validation. The removal of the old _host.py and associated infrastructure is consistent with the PR's stated direction. The samples have appropriate production-readiness disclaimers documenting the need for auth and session partitioning. No blocking security or reliability issues found. The lock file changes are mechanical sys_platform marker additions.

✓ Test Coverage

Test coverage for the new helper-first API is solid on happy paths (parsing, session continuity, round-trip, state holders). Two gaps stand out: (1) the streaming helper responses_from_streaming_run has no error-path test — the old channel tested response.failed emission on mid-stream exceptions, but the new helper propagates exceptions without any test verifying that behavior or showing the app-level pattern; (2) the _model_from_update logic that captures model ids from raw streaming chunks and overrides the finalized response model is entirely untested. The PR replaces the old ResponsesChannel (with its comprehensive test suite) with standalone helper functions and new tests. The new test coverage adequately exercises the happy paths for parsing, session continuity, and streaming. However, there is a notable gap: responses_from_streaming_run has no error handling for mid-stream failures, and unlike the old test_channel.py which had test_sse_emits_failed_when_stream_raises verifying a response.failed event, the new tests include no equivalent coverage for error propagation during streaming. The hosting package's API surface change from channel-based to state-helper-based is well tested for core paths (SessionStore CRUD, target resolution, session creation). The Telegram channel deletion is intentional and its tests were appropriately co-deleted. Minor test coverage asymetry exists: WorkflowState lacks parity tests for cache_target=False and bare-awaitable rejection that AgentState has, though the underlying code paths are nearly identical. The new _state.py module (AgentState, WorkflowState, SessionStore) has solid test coverage in test_state.py covering all major paths: CRUD operations with validation, target resolution from instances/callables/awaitables/builders, session create-once semantics, and session aliasing. The TypedDicts (AgentRunArgs, WorkflowRunArgs) are exercised through the hosting-responses helpers tested in test_parsing.py and test_http_round_trip.py. One minor gap: the RuntimeError guardrail on the synchronous .target property (raised when callable/awaitable targets haven't been resolved yet) lacks a direct test. The deleted test files (test_host.py, test_host_disk.py, test_isolation.py, test_types.py) covered production code that was also removed in this PR, so no coverage regression on existing code. The new test_state.py provides solid coverage of SessionStore, AgentState, and WorkflowState core behaviors. Minor coverage gap: the .target property's RuntimeError guard path (when accessed before async resolution) is untested.

✓ Failure Modes

The responses_from_streaming_run helper has no error handling around the stream iteration loop. If the agent's stream raises an exception mid-iteration, the SSE stream terminates without a terminal response.completed or response.failed event, violating the Responses SSE protocol contract. The old channel code (deleted in this PR) explicitly caught stream errors and emitted response.failed. Neither the sample nor the test fixture wrap the helper in try/except, confirming this is the intended (and broken) usage pattern. The new _state.py introduces AgentState.get_or_create_session with a get-check-create-set pattern that lacks the atomic protection the deleted _host.py explicitly documented. The old code used dict.setdefault to prevent concurrent calers from silently overwriting each other's sessions. While the default in-memory SessionStore with a pre-resolved target has no real async yield points (so no race in CPython today), the class is designed for subclassing with real async I/O backends (Redis, databases — mentioned in the SessionStore docstring), where the race becomes real.

✗ Design Approach

I did not find another design-level issue in the shown changes that met the evidence bar. The new helper-first parsing/rendering split is moving in the right direction, but the streaming helper currently drops important parts of the Responses SSE contract: it does not emit a terminal failure event when the upstream stream raises, and it only streams plain text deltas, so reasoning/tool/media updates appear only in the final envelope. As written, the package no longer exposes the prior host/channel API, which conflicts with the PR rationale that the helper/state surface is being introduced while the existing hosting code remains available. The new helper-first state surface is mostly coherent, but AgentState.get_or_create_session dropped the old host's concurrency protection. That makes concurrent requests for the same app-selected session id race to create separate AgentSession objects, so one request can run against a session instance that is no longer the stored canonical session for later turns. The new helper-first state surface is mostly internally consistent, but the SessionStore design currently locks in a mismatch between its documented continuation semantics and the behavior the tests endorse. In particular, earlier response IDs are documented as independently resumable, yet the new tests explicitly alias multiple IDs to the same mutable AgentSession object, which means older IDs will silently resume from the latest mutated state instead of their original continuation point. I found two design-level problems in the new workflow sample. First, the route caches one Workflow instance for all requests even though the core workflow runtime explicitly forbids concurrent runs on the same instance, so overlapping /responses calls will fail instead of behaving like independent conversations. Second, the sample records response_id -> checkpoint_id by asking a shared checkpoint directory for the workflow’s global latest checkpoint, which can bind one response id to another conversation’s checkpoint under concurrent traffic; the repo already uses per-context checkpoint storage to avoid that class of mix-up.

Flagged Issues

• python/packages/hosting/tests/hosting/test_state.py:116-124 enshrines same-object aliasing for multiple response IDs, but _state.py and README.md promise calers can continue from any earlier point in a conversation. Because AgentSession is a mutable state container, this approach cannot preserve historical continuation points — either snapshot/clone per stored id, or narrow the docs/tests to no longer promise arbitrary historical resume.
• python/samples/04-hosting/af-hosting/local_responses_workflow/app.py:210 caches a single workflow instance behind the FastAPI route, but Workflow.run() raises WorkflowException on overlapping executions. Two concurrent POST /responses calls will produce a 500. The route needs a fresh workflow per request (with a stable explicit workflow name for checkpoint compatibility).
• python/samples/04-hosting/af-hosting/local_responses_workflow/app.py:237 resolves the cursor with get_latest(workflow_name=target.name) from one shared checkpoint directory. Since FileCheckpointStorage.get_latest() is only scoped by workflow name, different conversations for the same workflow can race and alias the wrong checkpoint. Scope checkpoint storage per conversation/response id, or capture the checkpoint produced by the specific run.

---

Automated review by eavanvalkenburg's agents

[github-actions]

Flagged issue

python/packages/hosting/tests/hosting/test_state.py:116-124 enshrines same-object aliasing for multiple response IDs, but _state.py and README.md promise calers can continue from any earlier point in a conversation. Because AgentSession is a mutable state container, this approach cannot preserve historical continuation points — either snapshot/clone per stored id, or narrow the docs/tests to no longer promise arbitrary historical resume.

---

Source: automated DevFlow PR review

[github-actions]

Flagged issue

python/samples/04-hosting/af-hosting/local_responses_workflow/app.py:210 caches a single workflow instance behind the FastAPI route, but Workflow.run() raises WorkflowException on overlapping executions. Two concurrent POST /responses calls will produce a 500. The route needs a fresh workflow per request (with a stable explicit workflow name for checkpoint compatibility).

---

Source: automated DevFlow PR review

[github-actions]

Flagged issue

python/samples/04-hosting/af-hosting/local_responses_workflow/app.py:237 resolves the cursor with get_latest(workflow_name=target.name) from one shared checkpoint directory. Since FileCheckpointStorage.get_latest() is only scoped by workflow name, different conversations for the same workflow can race and alias the wrong checkpoint. Scope checkpoint storage per conversation/response id, or capture the checkpoint produced by the specific run.

---

Source: automated DevFlow PR review

[moonbox3]

One conflicted file, @eavanvalkenburg