Function handler

sdk-python/copilotkit/integrations/fastapi.py:77–170

Handle FastAPI request

(request: Request, sdk: CopilotKitRemoteEndpoint)

async def handler(request: Request, sdk: CopilotKitRemoteEndpoint):
    """Handle FastAPI request"""

    try:
        body = await request.json()
    except:  # pylint: disable=bare-except
        body = None

    # Propagate x-aimock-* headers to outgoing LLM calls via ContextVar
    set_forwarded_headers(dict(request.headers))

    path = request.path_params.get("path")
    method = request.method
    context = cast(
        CopilotKitContext,
        {
            "properties": (body or {}).get("properties", {}),
            "frontend_url": (body or {}).get("frontendUrl", None),
            "headers": request.headers,
        },
    )

    # handle / request for info endpoint
    if method in ["GET", "POST"] and path == "":
        accept_header = request.headers.get("accept", "")
        return await handle_info(
            sdk=sdk,
            context=context,
            as_html="text/html" in accept_header,
        )

    # handle /agent/name request for executing an agent
    if method == "POST" and (match := re.match(r"agent/([a-zA-Z0-9_-]+)", path)):
        name = match.group(1)
        body = body or {}

        thread_id = body.get("threadId", str(uuid.uuid4()))
        state = body.get("state", {})
        messages = body.get("messages", [])
        actions = body.get("actions", [])

        # used for LangGraph only
        node_name = body.get("nodeName")

        return handle_execute_agent(
            sdk=sdk,
            context=context,
            thread_id=thread_id,
            node_name=node_name,
            name=name,
            state=state,
            messages=messages,
            actions=actions,
        )

    # handle /agent/name/state request for getting agent state
    if method == "POST" and (match := re.match(r"agent/([a-zA-Z0-9_-]+)/state", path)):
        name = match.group(1)