Is it possible to implement add_agent_framework_fastapi_endpoint with OBO?

[djw-bsn]

We are trying to connect to a V2 Foundry Agent and implement a fastapi endpoint. Is it possible to implement add_agent_framework_fastapi_endpoint with On Behalf Of (OBO) credentials? As far as I can see it has to be a fixed credential but maybe that is a limitation of AIProjectClient (azure.ai.projects) and not AzureAIClient or the add_agent_framework_fastapi_endpoint from this project?

OnBehalfOfCredential needs a user_assertion (token) from the request but not sure how we can access this with add_agent_framework_fastapi_endpoint?

    OnBehalfOfCredential(
        client_id=OBO_CLIENT_ID,
        client_secret=OBO_CLIENT_SECRET,
        tenant_id=OBO_TENANT_ID,
        user_assertion=user_token,
    )

Working example with cli creds:

from azure.identity.aio import AzureCliCredential
from azure.ai.projects.aio import AIProjectClient

from agent_framework import ChatAgent
from agent_framework.azure import AzureAIClient
from agent_framework_ag_ui import add_agent_framework_fastapi_endpoint

.....

    project_client = AIProjectClient(
        endpoint=PROJECT_ENDPOINT,
        credential=AzureCliCredential(),
    )

    chat_client = AzureAIClient(
        project_client=project_client,
        agent_name=FOUNDRY_AGENT_NAME,
        use_latest_version=True,
    )

    app_agent = ChatAgent(
        chat_client=chat_client,
        instructions="You are a helpful assistant powered by a Foundry v2 agent.",
    )

    add_agent_framework_fastapi_endpoint(
        app,
        app_agent,
        "/chat",
    )

.....

[moonbox3]

Hi @djw-bsn, thanks for your question.

add_agent_framework_fastapi_endpoint doesn't support OBO out of the box. The agent is instantiated once at startup with a fixed credential, but OBO requires a per-request credential since the user_assertion changes per user.

To move forward with OBO, one would need to create a custom endpoint that extracts the bearer token and builds the credential per-request. Something like:

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from azure.identity.aio import OnBehalfOfCredential
from agent_framework import ChatAgent
from agent_framework.azure import AzureAIClient
from agent_framework.ag_ui import AgentFrameworkAgent
from ag_ui.encoder import EventEncoder

app = FastAPI()

@app.post("/chat")
async def chat_endpoint(request: Request):
    user_token = request.headers.get("Authorization", "")[7:]  # strip "Bearer "
    
    credential = OnBehalfOfCredential(
        client_id=OBO_CLIENT_ID, client_secret=OBO_CLIENT_SECRET,
        tenant_id=OBO_TENANT_ID, user_assertion=user_token,
    )
    chat_client = AzureAIClient(
        project_endpoint=PROJECT_ENDPOINT, credential=credential,
        agent_name=FOUNDRY_AGENT_NAME, use_latest_version=True,
    )
    agent = ChatAgent(chat_client=chat_client, instructions="You are a helpful assistant.")
    wrapped = AgentFrameworkAgent(agent=agent)
    
    async def stream():
        encoder = EventEncoder()
        try:
            async for event in wrapped.run_agent(await request.json()):
                yield encoder.encode(event)
        finally:
            await chat_client.close()
    
    return StreamingResponse(stream(), media_type="text/event-stream")

Do note that the tradeoff of doing it this way is overhead given that there is per-request client creation.

[djw-bsn]

Thanks @moonbox3 for the quick response. I figured as much but wanted to be sure. I had thought about something like this and creating an agent per user/thread but I was worried about losing the AG-UI support. But seems like your example here also keeps the AG-UI support so I will definitely give it a try. Thanks!