From 2444895ca43a47cddfab1fd7330a92c6e2582302 Mon Sep 17 00:00:00 2001 From: Vini Brasil Date: Fri, 12 Jun 2026 19:47:58 -0700 Subject: [PATCH] Implement Flow definition run tools without Python code (#6144) A `do:` step can now say `call: tool` and name a CrewAI tool to run, passing its inputs under `with:`. Before this, a definition could only point at Python code to run. ```yaml methods: search: start: true do: call: tool ref: crewai_tools:ExaSearchTool with: search_query: ai agents ``` --- lib/crewai/src/crewai/flow/dsl/_utils.py | 3 +- lib/crewai/src/crewai/flow/flow_definition.py | 21 +++- .../src/crewai/flow/runtime/_resolvers.py | 37 ++++++- lib/crewai/tests/test_flow_definition.py | 2 + lib/crewai/tests/test_flow_from_definition.py | 97 +++++++++++++++++++ 5 files changed, 155 insertions(+), 5 deletions(-) diff --git a/lib/crewai/src/crewai/flow/dsl/_utils.py b/lib/crewai/src/crewai/flow/dsl/_utils.py index cdd7bdbca..b203bcd62 100644 --- a/lib/crewai/src/crewai/flow/dsl/_utils.py +++ b/lib/crewai/src/crewai/flow/dsl/_utils.py @@ -9,6 +9,7 @@ from typing_extensions import TypeIs from crewai.flow.flow_definition import ( FlowActionDefinition, + FlowCodeActionDefinition, FlowConfigDefinition, FlowConversationalDefinition, FlowConversationalRouterDefinition, @@ -83,7 +84,7 @@ def _stamp_inherited_conversational_metadata( def _method_action(method: Any) -> FlowActionDefinition: - return FlowActionDefinition(ref=f"{method.__module__}:{method.__qualname__}") + return FlowCodeActionDefinition(ref=f"{method.__module__}:{method.__qualname__}") def _set_flow_method_definition( diff --git a/lib/crewai/src/crewai/flow/flow_definition.py b/lib/crewai/src/crewai/flow/flow_definition.py index e55bb2bf7..c1fd8afcf 100644 --- a/lib/crewai/src/crewai/flow/flow_definition.py +++ b/lib/crewai/src/crewai/flow/flow_definition.py @@ -28,6 +28,7 @@ FlowDefinitionCondition = str | dict[str, Any] __all__ = [ "FlowActionDefinition", + "FlowCodeActionDefinition", "FlowConfigDefinition", "FlowConversationalDefinition", "FlowConversationalRouterDefinition", @@ -38,6 +39,7 @@ __all__ = [ "FlowMethodDefinition", "FlowPersistenceDefinition", "FlowStateDefinition", + "FlowToolActionDefinition", ] @@ -142,13 +144,28 @@ class FlowHumanFeedbackDefinition(BaseModel): return _object_ref(value) -class FlowActionDefinition(BaseModel): - """What a Flow method node executes, independent of when it fires.""" +class FlowCodeActionDefinition(BaseModel): + """A Flow method action that executes importable Python code.""" + + model_config = ConfigDict(extra="forbid") call: TypingLiteral["code"] = "code" ref: str +class FlowToolActionDefinition(BaseModel): + """A Flow method action that invokes a CrewAI tool.""" + + model_config = ConfigDict(populate_by_name=True, extra="forbid") + + call: TypingLiteral["tool"] + ref: str + with_: dict[str, Any] | None = Field(default=None, alias="with") + + +FlowActionDefinition = FlowCodeActionDefinition | FlowToolActionDefinition + + class FlowMethodDefinition(BaseModel): """Static definition of one Flow method and its execution roles.""" diff --git a/lib/crewai/src/crewai/flow/runtime/_resolvers.py b/lib/crewai/src/crewai/flow/runtime/_resolvers.py index 063009d4f..be886f6ff 100644 --- a/lib/crewai/src/crewai/flow/runtime/_resolvers.py +++ b/lib/crewai/src/crewai/flow/runtime/_resolvers.py @@ -13,7 +13,11 @@ import inspect from operator import attrgetter from typing import TYPE_CHECKING, Any, cast -from crewai.flow.flow_definition import FlowActionDefinition +from crewai.flow.flow_definition import ( + FlowActionDefinition, + FlowCodeActionDefinition, + FlowToolActionDefinition, +) if TYPE_CHECKING: @@ -51,7 +55,7 @@ def resolve_instance_ref(ref: str, *, field: str) -> Any: def _resolve_code_action( - flow: Flow[Any], action: FlowActionDefinition + flow: Flow[Any], action: FlowCodeActionDefinition ) -> Callable[..., Any]: ref = action.ref target = resolve_ref(ref, field="do") @@ -63,8 +67,37 @@ def _resolve_code_action( return handler +def _resolve_tool_action( + _flow: Flow[Any], action: FlowToolActionDefinition +) -> Callable[..., Any]: + target = resolve_ref(action.ref, field="do") + from crewai.tools import BaseTool + + if not (inspect.isclass(target) and issubclass(target, BaseTool)): + raise InvalidRefError( + f"invalid tool ref {action.ref!r}; expected a BaseTool class" + ) + + try: + tool_cls = cast(Callable[[], BaseTool], target) + tool = tool_cls() + except Exception as e: + raise InvalidRefError( + f"cannot instantiate tool ref {action.ref!r} without arguments: {e}" + ) from e + + tool_kwargs = action.with_ or {} + + def run_tool(*_args: Any, **_kwargs: Any) -> Any: + return tool.run(**tool_kwargs) + + return run_tool + + def resolve_action(flow: Flow[Any], action: FlowActionDefinition) -> Callable[..., Any]: """Turn one `do:` action into the callable the flow runs for that node.""" if action.call == "code": return _resolve_code_action(flow, action) + if action.call == "tool": + return _resolve_tool_action(flow, action) raise ValueError(f"unknown call type {action.call!r}") diff --git a/lib/crewai/tests/test_flow_definition.py b/lib/crewai/tests/test_flow_definition.py index ebbb2e60d..a1bda89a5 100644 --- a/lib/crewai/tests/test_flow_definition.py +++ b/lib/crewai/tests/test_flow_definition.py @@ -37,6 +37,7 @@ def test_flow_public_exports_are_explicit(): } assert set(flow_definition.__all__) == { "FlowActionDefinition", + "FlowCodeActionDefinition", "FlowConfigDefinition", "FlowConversationalDefinition", "FlowConversationalRouterDefinition", @@ -47,6 +48,7 @@ def test_flow_public_exports_are_explicit(): "FlowMethodDefinition", "FlowPersistenceDefinition", "FlowStateDefinition", + "FlowToolActionDefinition", } assert "build_flow_structure" in flow_visualization.__all__ assert "calculate_node_levels" not in flow_visualization.__all__ diff --git a/lib/crewai/tests/test_flow_from_definition.py b/lib/crewai/tests/test_flow_from_definition.py index 5175623a8..df0a8d641 100644 --- a/lib/crewai/tests/test_flow_from_definition.py +++ b/lib/crewai/tests/test_flow_from_definition.py @@ -24,9 +24,18 @@ from crewai.flow.flow_definition import FlowConfigDefinition, FlowDefinition from crewai.flow.persistence import persist from crewai.flow.persistence.base import FlowPersistence from crewai.state.checkpoint_config import CheckpointConfig +from crewai.tools import BaseTool from crewai.types.streaming import FlowStreamingOutput +class StaticSearchTool(BaseTool): + name: str = "StaticSearchTool" + description: str = "Returns a deterministic search result." + + def _run(self, search_query: str, prefix: str = "search") -> str: + return f"{prefix}:{search_query}" + + class ChainFlow(Flow): @start() def begin(self): @@ -490,6 +499,94 @@ def test_flow_definition_stamps_refs(): assert definition.methods["shout"].do.ref == f"{__name__}:ChainFlow.shout" +def test_from_definition_runs_tool_action_with_static_inputs(): + yaml_str = f""" +schema: crewai.flow/v1 +name: ToolFlow +methods: + search: + do: + call: tool + ref: {__name__}:StaticSearchTool + with: + search_query: ai agents + prefix: found + start: true +""" + + flow = Flow.from_definition(FlowDefinition.from_yaml(yaml_str)) + + assert flow.kickoff() == "found:ai agents" + + +def test_tool_action_round_trips_with_inputs(): + definition = FlowDefinition.from_dict( + { + "schema": "crewai.flow/v1", + "name": "ToolFlow", + "methods": { + "search": { + "start": True, + "do": { + "call": "tool", + "ref": f"{__name__}:StaticSearchTool", + "with": {"search_query": "ai agents"}, + }, + } + }, + } + ) + + assert definition.to_dict()["methods"]["search"]["do"] == { + "call": "tool", + "ref": f"{__name__}:StaticSearchTool", + "with": {"search_query": "ai agents"}, + } + assert Flow.from_definition(definition).kickoff() == "search:ai agents" + + +def test_tool_action_requires_module_qualname_ref(): + definition = FlowDefinition.from_dict( + { + "schema": "crewai.flow/v1", + "name": "ToolFlow", + "methods": { + "search": { + "start": True, + "do": { + "call": "tool", + "ref": f"{__name__}.StaticSearchTool", + "with": {"search_query": "ai agents"}, + }, + } + }, + } + ) + + with pytest.raises(ValueError, match="expected 'module:qualname'"): + Flow.from_definition(definition) + + +def test_code_action_rejects_tool_inputs(): + with pytest.raises(ValidationError): + FlowDefinition.from_dict( + { + "schema": "crewai.flow/v1", + "name": "InvalidCodeActionFlow", + "methods": { + "begin": { + "start": True, + "do": { + "call": "code", + "ref": f"{__name__}:ChainFlow.begin", + "with": {"search_query": "ai agents"}, + }, + } + }, + } + ) + + def test_pydantic_state_from_ref_parity(): flow, result = assert_parity(PydanticStateFlow, PYDANTIC_STATE_YAML) assert result == "count=1"