FastAPIHelper Integration

Applies to: 4.0.8+

FastAPIHelper is Agently's FastAPI integration layer. It turns a response provider (Agent / Request / Flow) into HTTP endpoints with a unified contract.

Good fit when you need:

• fast API delivery for AI features
• both non-streaming and streaming endpoints (SSE/WS)
• consistent input/output/error wrapping across providers

1. Supported response_provider types

FastAPIHelper(response_provider=...) supports:

• BaseAgent
• ModelRequest
• TriggerFlow
• TriggerFlowExecution
• custom Generator / AsyncGenerator functions

This makes progressive migration straightforward:

• start with Agent for speed
• move to TriggerFlow for orchestration
• keep the same external API contract

2. Full server example (with startup)

import os
import uvicorn

from agently import Agently
from agently.integrations.fastapi import FastAPIHelper

OLLAMA_BASE_URL = os.environ.get("OLLAMA_BASE_URL", "http://127.0.0.1:11434/v1")

Agently.set_settings(
    "OpenAICompatible",
    {
        "base_url": OLLAMA_BASE_URL,
        "model": "qwen2.5:7b",
        "options": {"temperature": 0.3},
    },
)

agent = Agently.create_agent()
agent.role("You are a concise and helpful assistant.", always=True)

app = FastAPIHelper(response_provider=agent)

(
    app
    .use_post("/agent/chat")
    .use_get("/agent/chat/get")
    .use_sse("/agent/chat/sse")
    .use_websocket("/agent/chat/ws")
)

@app.get("/health")
async def health():
    return {"ok": True, "provider": "agent", "model": "qwen2.5:7b"}

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8000)

3. Request and response contract

3.1 Request payload shape

{
  "data": { "input": "hello" },
  "options": {}
}

• data: provider input payload
• options: optional runtime options forwarded to provider methods

3.2 Default response wrapper

Success:

{ "status": 200, "data": "...", "msg": null }

Failure:

{
  "status": 422,
  "data": null,
  "msg": "Invalid request payload ...",
  "error": {
    "type": "ValueError",
    "module": "builtins",
    "message": "..."
  }
}

Default status mapping:

• ValueError -> 422
• TimeoutError -> 504
• all other exceptions -> 400

You can override this contract with response_warper.

4. Endpoint styles

Method	Example path	Notes
POST	/agent/chat	JSON body input, most common
GET	/agent/chat/get	JSON string via query payload
SSE	/agent/chat/sse	stream events via data: frames
WS	/agent/chat/ws	full duplex realtime channel

5. cURL testing

5.1 POST

curl -sS -X POST "http://127.0.0.1:8000/agent/chat" \
  -H "Content-Type: application/json" \
  -d '{
    "data": {"input": "Describe Agently in one sentence."},
    "options": {}
  }'

5.2 SSE

curl -N -G "http://127.0.0.1:8000/agent/chat/sse" \
  --data-urlencode 'payload={"data":{"input":"Count from 1 to 3"},"options":{}}'

6. TriggerFlow concurrency bridging

When provider is TriggerFlowExecution and request options includes concurrency, FastAPIHelper forwards it to execution.set_concurrency(...).

{
  "data": {"input": "..."},
  "options": {"concurrency": 2}
}

7. Production checklist

• enforce auth/rate limit/audit at API gateway
• validate request payload using FastAPI/Pydantic models
• configure timeout/retry/reconnect for SSE/WS clients
• preserve error.type + trace id for incident debugging
• use TriggerFlow runtime stream for long-running orchestration visibility

8. Reference examples

• examples/fastapi/fastapi_helper_agent_ollama.py
• examples/fastapi/fastapi_helper_triggerflow_ollama.py
• examples/fastapi/fastapi_helper_agent_request.py