Method execute_agent

sdk-python/copilotkit/sdk.py:297–344

Execute an agent

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

            raise ActionExecutionException(name, error) from error

    def execute_agent(  # pylint: disable=too-many-arguments
        self,
        *,
        context: CopilotKitContext,
        name: str,
        thread_id: str,
        state: dict,
        config: Optional[dict] = None,
        messages: List[Message],
        actions: List[ActionDict],
        node_name: str,
        meta_events: Optional[List[MetaEvent]] = None,
    ) -> Any:
        """
        Execute an agent
        """
        agents = self.agents(context) if callable(self.agents) else self.agents
        agent = next((agent for agent in agents if agent.name == name), None)
        if agent is None:
            raise AgentNotFoundException(name)

        self._log_request_info(
            title="Handling execute agent request:",
            data=[
                ("Context", context),
                ("Agent", agent.dict_repr()),
                ("Thread ID", thread_id),
                ("Node Name", node_name),
                ("State", state),
                ("Config", config),
                ("Messages", messages),
                ("Actions", actions),
                ("MetaEvents", meta_events),
            ],
        )

        try:
            return agent.execute(
                thread_id=thread_id,
                node_name=node_name,
                state=state,
                config=config,
                messages=messages,
                actions=actions,
                meta_events=meta_events,
            )
        except Exception as error:
            raise AgentExecutionException(name, error) from error

    async def get_agent_state(
        self,