Function handle_execute_agent

sdk-python/copilotkit/integrations/fastapi.py:274–309

Handle continue agent execution request with FastAPI

(  # pylint: disable=too-many-arguments
    *,
    sdk: CopilotKitRemoteEndpoint,
    context: CopilotKitContext,
    thread_id: str,
    name: str,
    state: dict,
    config: Optional[dict] = None,
    messages: List[Message],
    actions: List[ActionDict],
    node_name: str,
    meta_events: Optional[List[MetaEvent]] = None,
)

def handle_execute_agent(  # pylint: disable=too-many-arguments
    *,
    sdk: CopilotKitRemoteEndpoint,
    context: CopilotKitContext,
    thread_id: str,
    name: str,
    state: dict,
    config: Optional[dict] = None,
    messages: List[Message],
    actions: List[ActionDict],
    node_name: str,
    meta_events: Optional[List[MetaEvent]] = None,
):
    """Handle continue agent execution request with FastAPI"""
    try:
        events = sdk.execute_agent(
            context=context,
            thread_id=thread_id,
            name=name,
            node_name=node_name,
            state=state,
            config=config,
            messages=messages,
            actions=actions,
            meta_events=meta_events,
        )
        return StreamingResponse(events, media_type="application/json")
    except AgentNotFoundException as exc:
        logger.error("Agent not found: %s", exc, exc_info=True)
        return JSONResponse(content={"error": str(exc)}, status_code=404)
    except AgentExecutionException as exc:
        logger.error("Agent execution error: %s", exc, exc_info=True)
        return JSONResponse(content={"error": str(exc)}, status_code=500)
    except Exception as exc:  # pylint: disable=broad-except
        logger.error("Agent execution error: %s", exc, exc_info=True)
        return JSONResponse(content={"error": str(exc)}, status_code=500)


async def handle_get_agent_state(