Samodzielnie hostowane endpointy Responses OpenAI

Użyj agent-framework-hosting-responses, aby konwertować żądania i odpowiedzi w formacie OpenAI Responses w punkcie końcowym należącym do Twojej aplikacji. Serwer wybiera framework webowy, ścieżkę, uwierzytelnianie, autoryzację, opcje żądania i pamięć sesji.

pip install --pre agent-framework agent-framework-foundry agent-framework-hosting agent-framework-hosting-responses azure-identity

Przykład FastAPI to jedna implementacja. Ci sami pomocnicy pracują z platformami Django, Flask, Starlette, Azure Functions lub inną strukturą.

Hostowanie punktu końcowego agenta

Ten przykład konwertuje żądanie na wartości uruchomienia Agent Framework, stosuje zdefiniowaną przez aplikację listę dozwolonych opcji i zapisuje zaktualizowaną sesję pod nowo utworzonym identyfikatorem odpowiedzi.

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 określa element docelowy oraz ładuje lub tworzy sesję. Zapisz sesję po uruchomieniu lub po zakończeniu przebiegu przesyłania strumieniowego, ponieważ przebieg go aktualizuje.

Pełną aplikację, w tym definicję agenta i listę dozwolonych opcji żądania, można znaleźć w lokalnym przykładzie Responses.

Omówienie konwersji użycia odpowiedzi

W przypadku odpowiedzi agenta i przepływu roboczego pakiet hostingu zachowuje natywny obiekt OpenAI ResponseUsage zgodny z SDK w niezmienionej postaci, jeśli jest dostępny. Nie scala ona użycia natywnych odpowiedzi z programem Agent Framework UsageDetails.

Gdy użycie natywne nie jest dostępne, pakiet może odtworzyć użycie odpowiedzi z tych semantycznie pasujących pól struktury agentów:

Jawne wartości zerowe są zachowywane. Jeśli brakuje total_tokens, a liczba danych wejściowych i wyjściowych jest podana, pakiet oblicza ją jako sumę liczby danych wejściowych i wyjściowych.

Jeśli dostępne użycie platformy agentów jest niekompletne lub semantycznie niespójne ze schematem Odpowiedzi, pakiet pomija użycie. Nie zgaduje, nie kopiuje jednego licznika do drugiego ani nie sprawia, że odpowiedź, która w innym przypadku byłaby pomyślna, zakończy się niepowodzeniem. Źle sformułowana liczba skalarna pozostaje błędem.

Ta rekonstrukcja jest celowo stratna, ponieważ użycie Agent Framework jest niezależne od dostawcy, a użycie OpenAI Responses ma bogatszą, specyficzną dla dostawcy strukturę. Liczniki właściwe dla danego dostawcy, zgłaszane przez hostowanego agenta, takie jak dane użycia właściwe dla Anthropic, mogą zatem nie pojawić się w odpowiedzi otrzymanej przez aplikację wywołującą. Ta konwersja nie zapewnia współdziałania między różnymi wersjami zestawu OpenAI SDK uruchomionym w tym samym procesie.

Hostowanie punktu końcowego przepływu pracy

WorkflowState obsługuje wykonanie przepływu pracy, ale Twoja aplikacja odpowiada za przechowywanie punktów kontrolnych oraz mapowanie identyfikatora odpowiedzi na punkt kontrolny. Ten przykład przywraca punkt kontrolny wybrany przez autoryzowany previous_response_id, a następnie zapisuje kursor dla kolejnej odpowiedzi.

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,
        )
    )

Magazyn danych używany w przykładzie, oparty na plikach, służy do lokalnego programowania. Użyj trwałej pamięci masowej, gdy repliki mogą zostać ponownie uruchomione lub skalowane w poziomie.

Aby uzyskać szerszy format przewodu, zobacz Punkty końcowe zgodne z protokołem OpenAI.

Następne kroki

Głębiej:

• Omówienie samodzielnego hostingu
• A2A
• MCP
• Agenci hostowani Foundry