Function handler_v1

sdk-python/copilotkit/integrations/fastapi.py:173–234

Handle FastAPI request for v1

(
    sdk: CopilotKitRemoteEndpoint,
    method: str,
    path: str,
    body: Any,
    context: CopilotKitContext,
)

async def handler_v1(
    sdk: CopilotKitRemoteEndpoint,
    method: str,
    path: str,
    body: Any,
    context: CopilotKitContext,
):
    """Handle FastAPI request for v1"""

    if body is None:
        raise HTTPException(status_code=400, detail="Request body is required")

    if method == "POST" and path == "info":
        return await handle_info(sdk=sdk, context=context)

    if method == "POST" and path == "actions/execute":
        name = body_get_or_raise(body, "name")
        arguments = body.get("arguments", {})

        return await handle_execute_action(
            sdk=sdk,
            context=context,
            name=name,
            arguments=arguments,
        )

    if method == "POST" and path == "agents/execute":
        thread_id = body.get("threadId")
        node_name = body.get("nodeName")
        config = body.get("config")

        name = body_get_or_raise(body, "name")
        state = body_get_or_raise(body, "state")
        messages = body_get_or_raise(body, "messages")
        actions = cast(List[ActionDict], body.get("actions", []))
        meta_events = cast(List[MetaEvent], body.get("metaEvents", []))

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

    if method == "POST" and path == "agents/state":
        thread_id = body_get_or_raise(body, "threadId")
        name = body_get_or_raise(body, "name")

        return await handle_get_agent_state(
            sdk=sdk,
            context=context,
            thread_id=thread_id,