Note
Access to this page requires authorization. You can try signing in or changing directories.
Access to this page requires authorization. You can try changing directories.
Note
Self-hosting helpers for OpenAI Responses endpoints in .NET are coming soon.
Note
Self-hosting helpers for OpenAI Responses endpoints are not currently available for Go.
Use agent-framework-hosting-responses to convert OpenAI Responses-shaped requests and responses at an endpoint your application owns. Your server chooses the web framework, route, authentication, authorization, request options, and session storage.
pip install --pre agent-framework agent-framework-foundry agent-framework-hosting agent-framework-hosting-responses azure-identity
The FastAPI sample is one implementation. The same helpers work with Django, Flask, Starlette, Azure Functions, or another framework.
Host an agent endpoint
This sample converts the request to Agent Framework run values, applies an application-defined option allowlist, and persists the updated session under the newly created response ID.
app = FastAPI()
state = AgentState(
create_agent,
session_store=FileSessionStore(SESSIONS_DIR / "snapshots"),
)
ALLOWED_REQUEST_OPTIONS = frozenset({"max_tokens", "reasoning"})
@app.post("/responses", response_model=None)
async def responses(body: dict[str, Any] = Body(...)) -> JSONResponse | StreamingResponse: # noqa: B008
"""Handle one OpenAI Responses-shaped request."""
try:
run = responses_to_run(body)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
session_id, is_conversation_id = responses_session_id(body)
conversation_id = session_id if is_conversation_id else None
response_id = create_response_id()
# App-specific policy: allow only the request options this route is willing
# to honor. This denies tools, tool_choice, deployment/persistence fields,
# and all other caller-supplied options by default. Your app decides which
# options are allowed, altered, or denied.
options = {key: value for key, value in run["options"].items() if key in ALLOWED_REQUEST_OPTIONS}
options["reasoning"] = {"effort": "medium", "summary": "auto"}
options_for_run = cast(Any, options)
target = await state.get_target()
lookup_id = session_id or response_id
# An unknown id supplied through `conversation` becomes a new session here. Production apps
# can choose to require a separate "create conversation" API instead.
session = await state.get_or_create_session(lookup_id)
if run["stream"]:
stream = target.run(
run["messages"],
stream=True,
session=session,
options=options_for_run,
)
if not isinstance(stream, ResponseStream):
raise HTTPException(status_code=500, detail="agent did not return a response stream")
async def stream_events() -> AsyncIterator[str]:
async for event in responses_from_streaming_run(
stream,
response_id=response_id,
conversation_id=conversation_id,
):
yield event
# `agent.run(..., stream=True)` updates the session while the stream
# is consumed/finalized. Persist the selected continuation only
# after finalization.
if conversation_id is not None:
# A stable conversation id is a mutable head. Apps must ensure
# only one caller advances it at a time; AgentState does not
# serialize concurrent runs for the same id.
await state.set_session(conversation_id, session)
else:
await state.set_session(response_id, session)
return StreamingResponse(
stream_events(),
media_type="text/event-stream",
)
result = await target.run(
run["messages"],
session=session,
options=options_for_run,
)
# `agent.run(...)` updates the session. Persist the selected continuation
# only after the run completes.
AgentState resolves the target and loads or creates a session. Save the session after the run, or after a streaming run finishes, because the run updates it.
For the complete application, including the agent definition and request-option allowlist, see the local Responses sample.
Understand response usage conversion
For agent and workflow responses, the hosting package preserves an SDK-valid native OpenAI ResponseUsage object unchanged when one is available. It doesn't merge native Responses usage with Agent Framework UsageDetails.
When native usage isn't available, the package can reconstruct Responses usage from these semantically matching Agent Framework fields:
| Usage value | Agent Framework field |
|---|---|
| Input tokens | input_token_count |
| Output tokens | output_token_count |
| Cache-read input tokens | cache_read_input_token_count |
| Cache-write input tokens | cache_creation_input_token_count |
| Reasoning output tokens | reasoning_output_token_count |
Explicit zero values are preserved. If total_tokens is absent while both input and output counts are present, the package derives it as input plus output.
If the available Agent Framework usage is incomplete or semantically inconsistent with the Responses schema, the package omits usage. It doesn't guess, copy one counter into another, or fail an otherwise successful response. A malformed scalar count remains an error.
This reconstruction is intentionally lossy because Agent Framework usage is provider-neutral and OpenAI Responses usage has a richer, provider-specific shape. Provider-specific counters reported by a hosted agent, such as Anthropic-specific usage, therefore might not appear in the response received by the calling application. This conversion doesn't provide interoperability between different versions of the OpenAI SDK running in the same process.
Host a workflow endpoint
WorkflowState resolves the workflow, but your application owns checkpoint storage and the mapping from a response ID to a checkpoint. This sample restores the checkpoint selected by an authorized previous_response_id, then saves a cursor for the next response.
app = FastAPI()
state = WorkflowState(workflow_builder, cache_target=False)
@app.post("/responses", response_model=None)
async def responses(body: dict[str, Any] = Body(...)) -> JSONResponse: # noqa: B008
"""Handle one OpenAI Responses-shaped request for the workflow."""
try:
run = responses_to_run(body)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
# This sample demonstrates only Responses `previous_response_id`
# continuation, so reject `conversation` instead of treating it as a
# checkpoint cursor.
previous_response_id, is_conversation_id = responses_session_id(body)
if is_conversation_id:
raise HTTPException(
status_code=400,
detail="This server supports previous_response_id continuation only; conversation is not implemented.",
)
response_id = create_response_id()
target = await state.get_target()
if previous_response_id and (checkpoint_cursor := checkpoint_cursor_store.get(previous_response_id)) is not None:
# Restore first. Workflow.run does not allow `message` and
# `checkpoint_id` in the same call.
await target.run(
checkpoint_id=checkpoint_cursor["checkpoint_id"],
checkpoint_storage=checkpoint_storage_for(checkpoint_cursor["storage_id"]),
)
storage_id = response_id
checkpoint_storage = checkpoint_storage_for(storage_id)
result = await target.run(
message=workflow_prompt_from_messages(run["messages"]),
checkpoint_storage=checkpoint_storage,
)
latest = await checkpoint_storage.get_latest(workflow_name=target.name)
if latest is not None:
# Responses `previous_response_id` can point to any response id. Store
# the current response id as the cursor for this workflow continuation.
cursor = CheckpointCursor(checkpoint_id=latest.checkpoint_id, storage_id=storage_id)
checkpoint_cursor_store.set_many({response_id: cursor})
return JSONResponse(
responses_from_run(
response_from_workflow_result(result),
response_id=response_id,
)
)
The sample's file-backed storage is for local development. Use durable storage when replicas can restart or scale out.
Important
Treat previous_response_id and conversation_id as untrusted input. Authenticate and authorize the caller before using either ID to load or save a session or checkpoint.
For the broader wire format, see OpenAI-compatible endpoints.
Next steps
Go deeper: