FastAPI integration

stacgis_ai.fastapi_integration provides ready-made routers. All of them are also re-exported from the package root.

Routers

Factory	Purpose
create_agent_router(agent_loop_factory, prefix="/agent")	Chat + streaming + HitL endpoints (needs a factory that returns an AgentLoop)
create_flow_router(prefix="/stacgis_ai")	Health, tool schemas, tool listing, single & batch execution
create_tools_router(prefix="/tools")	Minimal REST tool endpoints
create_websocket_endpoint(...)	WebSocket streaming variant
default_router	Pre-wired convenience router

Mounting the agent router

from fastapi import FastAPI
from stacgis_ai.fastapi_integration.routers import create_agent_router

# async factory: build + initialise your AgentLoop per request/app
async def get_agent_loop():
    ...
    loop = AgentLoop(mcp_client=client, llm=llm, config=AgentLoopConfig(max_steps=8))
    await loop.initialize()
    return loop

app = FastAPI(title="GeoAgent")
app.include_router(create_agent_router(get_agent_loop), prefix="/api/v1")

Endpoints

POST /agent/chat — blocking

// request
{
  "message": "Find 5 cafes near London and buffer them by 200 m",
  "geometry": { "type": "Feature", "geometry": { "type": "Point", "coordinates": [0, 51] } },
  "approval_timeout_s": 120
}

// response
{
  "response": "…",
  "total_steps": 4,
  "tools_used": ["query_osm_features", "generate_buffer_geojson"],
  "total_duration_ms": 8342.1,
  "was_truncated": false
}

POST /agent/stream — SSE streaming (multipart/form-data)

Form fields: message (required), geometry (optional GeoJSON string), approval_timeout_s.

Response: text/event-stream. Each event is a JSON step:

data: {"step_number": 1, "state": "thinking",  "thought": "…", "is_complete": false}
data: {"step_number": 1, "state": "acting",    "tool_name": "geocode_address",
       "tool_args": {"address": "London"}, "is_complete": false}
data: {"step_number": 1, "state": "observing", "tool_result": "{…}", "is_complete": false}
data: {"step_number": 2, "state": "responding","thought": "…", "is_complete": true}

Step state values: thinking, acting, observing, responding, intermediate_reasoning, error, awaiting_approval, approval_resolved, frontend_tool. The final event always has is_complete: true.

POST /agent/approval/{approval_id} — HitL

// request
{ "approved": true, "reason": null }

// response
{ "status": "resolved", "approval_id": "…", "approved": true, "reason": null }

GET /agent/approval/pending lists requests currently waiting for a decision. 404 is returned for unknown or already-resolved ids.

Flow router (create_flow_router)

Method	Path	Description
GET	/health	Health + circuit-breaker stats
GET	/tools/schema	LLM-ready tool schemas
GET	/tools	List tools (filter by category, tag, search)
GET	/tools/{name}	Single tool definition
POST	/execute	Execute one tool
POST	/execute/batch	Execute many tools (parallel, fail-fast options)

CORS

The example enables permissive CORS for local development:

app.add_middleware(CORSMiddleware, allow_origins=["*"],
                   allow_credentials=True, allow_methods=["*"], allow_headers=["*"])

caution

Tighten allow_origins before deploying.