From c2876c5d720eb37f2ff516ca00c859e2fbb5f0b3 Mon Sep 17 00:00:00 2001 From: ogabrielluiz Date: Thu, 25 Jun 2026 13:44:05 -0300 Subject: [PATCH] feat(a2a): add flow_type categorization and A2A flow columns Add an explicit flow_type enum (workflow|agent), a2a_enabled, and a2a_card_overrides to the Flow model via an additive migration following the access_type precedent. Surface the fields in the flow create/read/update schemas, add a flow_type filter to the list endpoint, and add a suggest_flow_type auto-detect helper (UI suggestion only, never the stored source of truth). This is F1, the foundation of the A2A protocol support epic. --- ...3_add_flow_type_and_a2a_columns_to_flow.py | 49 ++++++ src/backend/base/langflow/api/v1/flows.py | 5 + .../base/langflow/api/v1/flows_helpers.py | 3 + src/backend/base/langflow/helpers/flow.py | 37 ++++- .../services/database/models/flow/model.py | 37 ++++- .../unit/alembic/test_flow_type_migration.py | 40 +++++ src/backend/tests/unit/api/v1/test_flows.py | 140 ++++++++++++++++++ .../unit/helpers/test_suggest_flow_type.py | 55 +++++++ 8 files changed, 364 insertions(+), 2 deletions(-) create mode 100644 src/backend/base/langflow/alembic/versions/9f1d1d602aa3_add_flow_type_and_a2a_columns_to_flow.py create mode 100644 src/backend/tests/unit/alembic/test_flow_type_migration.py create mode 100644 src/backend/tests/unit/helpers/test_suggest_flow_type.py diff --git a/src/backend/base/langflow/alembic/versions/9f1d1d602aa3_add_flow_type_and_a2a_columns_to_flow.py b/src/backend/base/langflow/alembic/versions/9f1d1d602aa3_add_flow_type_and_a2a_columns_to_flow.py new file mode 100644 index 0000000000..d8276723dd --- /dev/null +++ b/src/backend/base/langflow/alembic/versions/9f1d1d602aa3_add_flow_type_and_a2a_columns_to_flow.py @@ -0,0 +1,49 @@ +"""add flow_type and a2a columns to flow + +Revision ID: 9f1d1d602aa3 +Revises: a1f4c9d27b30 +Create Date: 2026-06-25 12:26:09.624143 + +Phase: EXPAND +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op +from langflow.utils import migration + +# revision identifiers, used by Alembic. +revision: str = "9f1d1d602aa3" # pragma: allowlist secret +down_revision: str | None = "a1f4c9d27b30" # pragma: allowlist secret +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + conn = op.get_bind() + flow_type_enum = sa.Enum("workflow", "agent", name="flow_type_enum") + flow_type_enum.create(conn, checkfirst=True) + with op.batch_alter_table("flow", schema=None) as batch_op: + if not migration.column_exists(table_name="flow", column_name="flow_type", conn=conn): + batch_op.add_column( + sa.Column("flow_type", flow_type_enum, server_default=sa.text("'workflow'"), nullable=False) + ) + if not migration.column_exists(table_name="flow", column_name="a2a_enabled", conn=conn): + batch_op.add_column(sa.Column("a2a_enabled", sa.Boolean(), server_default=sa.false(), nullable=True)) + if not migration.column_exists(table_name="flow", column_name="a2a_card_overrides", conn=conn): + batch_op.add_column(sa.Column("a2a_card_overrides", sa.JSON(), nullable=True)) + + +def downgrade() -> None: + conn = op.get_bind() + with op.batch_alter_table("flow", schema=None) as batch_op: + if migration.column_exists(table_name="flow", column_name="a2a_card_overrides", conn=conn): + batch_op.drop_column("a2a_card_overrides") + if migration.column_exists(table_name="flow", column_name="a2a_enabled", conn=conn): + batch_op.drop_column("a2a_enabled") + if migration.column_exists(table_name="flow", column_name="flow_type", conn=conn): + batch_op.drop_column("flow_type") + + flow_type_enum = sa.Enum("workflow", "agent", name="flow_type_enum") + flow_type_enum.drop(conn, checkfirst=True) diff --git a/src/backend/base/langflow/api/v1/flows.py b/src/backend/base/langflow/api/v1/flows.py index 6fd0707e71..f6de9fe412 100644 --- a/src/backend/base/langflow/api/v1/flows.py +++ b/src/backend/base/langflow/api/v1/flows.py @@ -58,6 +58,7 @@ from langflow.services.database.models.flow.model import ( FlowCreate, FlowHeader, FlowRead, + FlowType, FlowUpdate, ) @@ -122,6 +123,7 @@ async def read_flows( components_only: bool = False, get_all: bool = True, folder_id: UUID | None = None, + flow_type: FlowType | None = None, params: Annotated[Params, Depends()], header_flows: bool = False, ): @@ -157,6 +159,9 @@ async def read_flows( if components_only: stmt = stmt.where(Flow.is_component == True) # noqa: E712 + if flow_type is not None: + stmt = stmt.where(Flow.flow_type == flow_type) + if get_all: flows = (await session.exec(stmt)).all() flows = validate_is_component(flows) diff --git a/src/backend/base/langflow/api/v1/flows_helpers.py b/src/backend/base/langflow/api/v1/flows_helpers.py index 6e9c5eb168..97afcbc8a7 100644 --- a/src/backend/base/langflow/api/v1/flows_helpers.py +++ b/src/backend/base/langflow/api/v1/flows_helpers.py @@ -130,6 +130,9 @@ _UPDATABLE_FLOW_FIELDS: frozenset[str] = frozenset( "gradient", "locked", "mcp_enabled", + "flow_type", + "a2a_enabled", + "a2a_card_overrides", "action_name", "action_description", "access_type", diff --git a/src/backend/base/langflow/helpers/flow.py b/src/backend/base/langflow/helpers/flow.py index 86f5612d2a..d245b12653 100644 --- a/src/backend/base/langflow/helpers/flow.py +++ b/src/backend/base/langflow/helpers/flow.py @@ -10,7 +10,7 @@ from sqlalchemy.orm import aliased from sqlmodel import asc, desc, select from langflow.schema.schema import INPUT_FIELD_NAME -from langflow.services.database.models.flow.model import Flow, FlowRead +from langflow.services.database.models.flow.model import Flow, FlowRead, FlowType from langflow.services.deps import get_settings_service, session_scope if TYPE_CHECKING: @@ -587,3 +587,38 @@ def json_schema_from_flow(flow: Flow) -> dict: } return {"type": "object", "properties": properties, "required": required} + + +def suggest_flow_type(flow_data: dict | None) -> FlowType: + """Suggest ``agent`` vs ``workflow`` for a flow based on its graph contents. + + Returns ``FlowType.AGENT`` if any node's component is (a subclass of) + ``LCAgentComponent``, else ``FlowType.WORKFLOW``. This is a UI default + suggestion only, never the stored source of truth, so it never raises: + any node it cannot resolve is skipped and the flow falls back to + ``WORKFLOW``. The node's class is recovered from its own stored source + (``node.data.node.template.code.value``) via ``eval_custom_component_code``, + which evaluates the class definition without instantiating or running it. + """ + from lfx.base.agents.agent import LCAgentComponent + from lfx.custom.eval import eval_custom_component_code + + nodes = (flow_data or {}).get("nodes") or [] + for node in nodes: + try: + code = node["data"]["node"]["template"]["code"]["value"] + except (KeyError, TypeError): + continue + if not code: + continue + try: + component_class = eval_custom_component_code(code) + except Exception: # noqa: BLE001 - a suggestion must never fail the caller + logger.debug("suggest_flow_type: skipping a node whose code could not be evaluated", exc_info=True) + continue + try: + if issubclass(component_class, LCAgentComponent): + return FlowType.AGENT + except TypeError: + continue + return FlowType.WORKFLOW diff --git a/src/backend/base/langflow/services/database/models/flow/model.py b/src/backend/base/langflow/services/database/models/flow/model.py index e8ad086f00..29c4b1942e 100644 --- a/src/backend/base/langflow/services/database/models/flow/model.py +++ b/src/backend/base/langflow/services/database/models/flow/model.py @@ -10,8 +10,8 @@ import emoji from emoji import purely_emoji from lfx.log.logger import logger from pydantic import BaseModel, ValidationInfo, field_serializer, field_validator +from sqlalchemy import Boolean, Text, UniqueConstraint, false, text from sqlalchemy import Enum as SQLEnum -from sqlalchemy import Text, UniqueConstraint, text from sqlmodel import JSON, Column, Field, Relationship, SQLModel from langflow.schema.data import Data @@ -46,6 +46,13 @@ class AccessTypeEnum(str, Enum): PUBLIC = "PUBLIC" +class FlowType(str, Enum): + # Extensible: new kinds can be added without a breaking change. Lowercase + # values so the stored string matches the wire value (see values_callable). + WORKFLOW = "workflow" + AGENT = "agent" + + class FlowBase(SQLModel): # Supresses warnings during migrations __mapper_args__ = {"confirm_deleted_rows": False} @@ -83,6 +90,29 @@ class FlowBase(SQLModel): server_default=text("'PRIVATE'"), ), ) + flow_type: FlowType = Field( + default=FlowType.WORKFLOW, + sa_column=Column( + SQLEnum( + FlowType, + name="flow_type_enum", + values_callable=lambda enum: [member.value for member in enum], + ), + nullable=False, + server_default=text("'workflow'"), + ), + description="Whether the flow is a plain workflow or an agent (publishable over A2A)", + ) + a2a_enabled: bool | None = Field( + default=False, + sa_column=Column(Boolean, nullable=True, server_default=false()), + description="Can be exposed as an A2A agent (only meaningful when flow_type=agent)", + ) + a2a_card_overrides: dict | None = Field( + default=None, + sa_column=Column(JSON, nullable=True), + description="User overrides for the generated A2A agent card (skill description, examples, tags)", + ) @field_validator("endpoint_name") @classmethod @@ -256,6 +286,8 @@ class FlowHeader(BaseModel): access_type: AccessTypeEnum | None = Field(None, description="The access type of the flow") tags: list[str] | None = Field(None, description="The tags of the flow") mcp_enabled: bool | None = Field(None, description="Flag indicating whether the flow is exposed in the MCP server") + flow_type: FlowType | None = Field(None, description="Whether the flow is a plain workflow or an agent") + a2a_enabled: bool | None = Field(None, description="Flag indicating whether the flow is exposed as an A2A agent") action_name: str | None = Field(None, description="The name of the action associated with the flow") action_description: str | None = Field(None, description="The description of the action associated with the flow") @@ -279,6 +311,9 @@ class FlowUpdate(SQLModel): action_name: str | None = None action_description: str | None = None access_type: AccessTypeEnum | None = None + flow_type: FlowType | None = None + a2a_enabled: bool | None = None + a2a_card_overrides: dict | None = None fs_path: str | None = None @field_validator("endpoint_name") diff --git a/src/backend/tests/unit/alembic/test_flow_type_migration.py b/src/backend/tests/unit/alembic/test_flow_type_migration.py new file mode 100644 index 0000000000..566f26d92f --- /dev/null +++ b/src/backend/tests/unit/alembic/test_flow_type_migration.py @@ -0,0 +1,40 @@ +"""Structure tests for the flow_type / a2a columns migration (9f1d1d602aa3). + +Covers the additive flow.flow_type, flow.a2a_enabled and flow.a2a_card_overrides +columns: forward to head, then rollback to the prior revision. Runs on sqlite +and (when configured) postgres. +""" + +from __future__ import annotations + +from alembic import command +from sqlalchemy import create_engine, inspect + +from .test_migration_execution import _engine_url, _make_alembic_cfg, db_url # noqa: F401 + +_PRIOR_REVISION = "a1f4c9d27b30" # pragma: allowlist secret +_A2A_COLUMNS = {"flow_type", "a2a_enabled", "a2a_card_overrides"} + + +def _flow_columns(db_url: str) -> set[str]: # noqa: F811 + engine = create_engine(_engine_url(db_url)) + try: + with engine.connect() as connection: + return {c["name"] for c in inspect(connection).get_columns("flow")} + finally: + engine.dispose() + + +def test_flow_has_flow_type_and_a2a_columns(db_url): # noqa: F811 + alembic_cfg = _make_alembic_cfg(db_url) + command.upgrade(alembic_cfg, "head") + + assert _flow_columns(db_url) >= _A2A_COLUMNS + + +def test_flow_type_columns_dropped_on_downgrade(db_url): # noqa: F811 + alembic_cfg = _make_alembic_cfg(db_url) + command.upgrade(alembic_cfg, "head") + command.downgrade(alembic_cfg, _PRIOR_REVISION) + + assert _A2A_COLUMNS.isdisjoint(_flow_columns(db_url)) diff --git a/src/backend/tests/unit/api/v1/test_flows.py b/src/backend/tests/unit/api/v1/test_flows.py index d143363219..deef1ea91c 100644 --- a/src/backend/tests/unit/api/v1/test_flows.py +++ b/src/backend/tests/unit/api/v1/test_flows.py @@ -276,6 +276,146 @@ async def test_patch_flow_updates_access_and_action_fields(client: AsyncClient, assert result["action_description"] == "Shared flow action" +async def test_create_flow_defaults_to_workflow_type(client: AsyncClient, logged_in_headers): + """A flow created without flow_type is a workflow with A2A off.""" + response = await client.post( + "api/v1/flows/", + json={"name": "default_type_flow", "data": {}}, + headers=logged_in_headers, + ) + assert response.status_code == status.HTTP_201_CREATED + result = response.json() + assert result["flow_type"] == "workflow" + assert result["a2a_enabled"] is False + assert result["a2a_card_overrides"] is None + + +async def test_create_agent_flow_round_trips(client: AsyncClient, logged_in_headers): + """flow_type=agent and the a2a fields persist through create and read.""" + create_response = await client.post( + "api/v1/flows/", + json={ + "name": "agent_flow", + "data": {}, + "flow_type": "agent", + "a2a_enabled": True, + "a2a_card_overrides": {"skill_description": "does things"}, + }, + headers=logged_in_headers, + ) + assert create_response.status_code == status.HTTP_201_CREATED + created = create_response.json() + assert created["flow_type"] == "agent" + assert created["a2a_enabled"] is True + + flow_id = created["id"] + read_response = await client.get(f"api/v1/flows/{flow_id}", headers=logged_in_headers) + assert read_response.status_code == status.HTTP_200_OK + read = read_response.json() + assert read["flow_type"] == "agent" + assert read["a2a_enabled"] is True + assert read["a2a_card_overrides"] == {"skill_description": "does things"} + + +async def test_patch_flow_updates_flow_type_and_a2a(client: AsyncClient, logged_in_headers): + """PATCH can promote a workflow to an agent and set the a2a fields.""" + create_response = await client.post( + "api/v1/flows/", + json={"name": "patch_flow_type_flow", "data": {}}, + headers=logged_in_headers, + ) + assert create_response.status_code == status.HTTP_201_CREATED + flow_id = create_response.json()["id"] + + response = await client.patch( + f"api/v1/flows/{flow_id}", + json={"flow_type": "agent", "a2a_enabled": True, "a2a_card_overrides": {"tags": ["x"]}}, + headers=logged_in_headers, + ) + assert response.status_code == status.HTTP_200_OK + result = response.json() + assert result["flow_type"] == "agent" + assert result["a2a_enabled"] is True + assert result["a2a_card_overrides"] == {"tags": ["x"]} + + +async def test_read_flows_filtered_by_flow_type(client: AsyncClient, logged_in_headers): + """The list endpoint filtered by flow_type=agent returns only agent flows.""" + workflow_response = await client.post( + "api/v1/flows/", + json={"name": "a_workflow_flow", "data": {}}, + headers=logged_in_headers, + ) + agent_response = await client.post( + "api/v1/flows/", + json={"name": "an_agent_flow", "data": {}, "flow_type": "agent"}, + headers=logged_in_headers, + ) + workflow_id = workflow_response.json()["id"] + agent_id = agent_response.json()["id"] + + response = await client.get( + "api/v1/flows/", + params={"get_all": True, "flow_type": "agent"}, + headers=logged_in_headers, + ) + assert response.status_code == status.HTTP_200_OK + result = response.json() + returned_ids = {flow["id"] for flow in result} + assert agent_id in returned_ids + assert workflow_id not in returned_ids + assert all(flow["flow_type"] == "agent" for flow in result) + + +async def test_create_agent_flow_defaults_a2a_disabled(client: AsyncClient, logged_in_headers): + """Creating an agent flow without a2a_enabled leaves A2A off by default.""" + response = await client.post( + "api/v1/flows/", + json={"name": "agent_no_a2a_flow", "data": {}, "flow_type": "agent"}, + headers=logged_in_headers, + ) + assert response.status_code == status.HTTP_201_CREATED + result = response.json() + assert result["flow_type"] == "agent" + assert result["a2a_enabled"] is False + + +async def test_read_flows_rejects_invalid_flow_type(client: AsyncClient, logged_in_headers): + """An unknown flow_type query value is rejected by enum validation.""" + response = await client.get( + "api/v1/flows/", + params={"get_all": True, "flow_type": "not_a_real_type"}, + headers=logged_in_headers, + ) + assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY + + +async def test_read_flows_header_mode_filtered_by_flow_type(client: AsyncClient, logged_in_headers): + """The flow_type filter also applies on the header_flows (compressed) list path.""" + await client.post( + "api/v1/flows/", + json={"name": "header_workflow_flow", "data": {}}, + headers=logged_in_headers, + ) + agent_response = await client.post( + "api/v1/flows/", + json={"name": "header_agent_flow", "data": {}, "flow_type": "agent"}, + headers=logged_in_headers, + ) + agent_id = agent_response.json()["id"] + + response = await client.get( + "api/v1/flows/", + params={"get_all": True, "header_flows": True, "flow_type": "agent"}, + headers=logged_in_headers, + ) + assert response.status_code == status.HTTP_200_OK + result = response.json() + returned_ids = {flow["id"] for flow in result} + assert agent_id in returned_ids + assert all(flow["flow_type"] == "agent" for flow in result) + + async def test_create_flows(client: AsyncClient, logged_in_headers): amount_flows = 10 basic_case = { diff --git a/src/backend/tests/unit/helpers/test_suggest_flow_type.py b/src/backend/tests/unit/helpers/test_suggest_flow_type.py new file mode 100644 index 0000000000..b2e90cdbf4 --- /dev/null +++ b/src/backend/tests/unit/helpers/test_suggest_flow_type.py @@ -0,0 +1,55 @@ +"""Tests for helpers.flow.suggest_flow_type (F1 agent auto-detect).""" + +from __future__ import annotations + +import json +from pathlib import Path + +import langflow +import pytest +from langflow.helpers.flow import suggest_flow_type +from langflow.services.database.models.flow.model import FlowType + +_STARTERS = Path(langflow.__file__).parent / "initial_setup" / "starter_projects" + + +def _load_agent_starter() -> dict: + """Return the graph data of a real starter project that contains an Agent node.""" + for path in sorted(_STARTERS.glob("*.json")): + data = json.loads(path.read_text(encoding="utf-8")).get("data") or {} + if any((node.get("data") or {}).get("type") == "Agent" for node in data.get("nodes") or []): + return data + pytest.skip("No starter project with an Agent node found") + return {} + + +def _first_non_agent_node(graph_data: dict) -> dict: + for node in graph_data.get("nodes") or []: + node_data = node.get("data") or {} + if node_data.get("type") != "Agent" and node_data.get("node", {}).get("template", {}).get("code", {}).get( + "value" + ): + return node + pytest.skip("No non-agent node with code found in starter") + return {} + + +def test_suggest_agent_for_flow_with_agent_component(): + agent_graph = _load_agent_starter() + assert suggest_flow_type(agent_graph) == FlowType.AGENT + + +def test_suggest_workflow_for_flow_without_agent_component(): + agent_graph = _load_agent_starter() + workflow_graph = {"nodes": [_first_non_agent_node(agent_graph)], "edges": []} + assert suggest_flow_type(workflow_graph) == FlowType.WORKFLOW + + +def test_suggest_workflow_for_empty_or_missing_data(): + assert suggest_flow_type({"nodes": [], "edges": []}) == FlowType.WORKFLOW + assert suggest_flow_type(None) == FlowType.WORKFLOW + + +def test_suggest_never_raises_on_malformed_nodes(): + bad = {"nodes": [{"data": {"node": {"template": {"code": {"value": "this is not python ("}}}}}]} + assert suggest_flow_type(bad) == FlowType.WORKFLOW