Run a node directly via NodeRunner without an orchestrator.
(
self,
node: BaseNode,
node_input: Any,
*,
use_as_output: bool = False,
run_id: str | None = None,
use_sub_branch: bool = False,
override_branch: str | None = None,
override_isolation_scope: str | None = None,
resume_inputs: dict[str, Any] | None = None,
)
| 899 | # ============================================================================ |
| 900 | |
| 901 | async def _run_node_standalone( |
| 902 | self, |
| 903 | node: BaseNode, |
| 904 | node_input: Any, |
| 905 | *, |
| 906 | use_as_output: bool = False, |
| 907 | run_id: str | None = None, |
| 908 | use_sub_branch: bool = False, |
| 909 | override_branch: str | None = None, |
| 910 | override_isolation_scope: str | None = None, |
| 911 | resume_inputs: dict[str, Any] | None = None, |
| 912 | ) -> Context: |
| 913 | """Run a node directly via NodeRunner without an orchestrator.""" |
| 914 | from ..workflow._node_runner import NodeRunner |
| 915 | |
| 916 | runner = NodeRunner( |
| 917 | node=node, |
| 918 | parent_ctx=self, |
| 919 | run_id=run_id, |
| 920 | use_as_output=use_as_output, |
| 921 | use_sub_branch=use_sub_branch, |
| 922 | override_branch=override_branch, |
| 923 | override_isolation_scope=override_isolation_scope, |
| 924 | ) |
| 925 | return await runner.run(node_input=node_input, resume_inputs=resume_inputs) |
no test coverage detected