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.

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)

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)

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",

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

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")

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))

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 = {

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