pengjunjie/langflow
1
0
Fork 1
mirror of https://github.com/langflow-ai/langflow.git synced 2026-07-25 20:20:08 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(a2a): PR prep — a2a-sdk dependency, Agent tool injection hook, ruff formatting

Browse Source 
Prepare the A2A implementation for PR readiness:

- Add a2a-sdk>=0.3.0 to langflow-base dependencies in pyproject.toml
- Add _inject_a2a_tools() hook to LCToolsAgentComponent in lfx:
  checks graph.context for A2A execution context, creates and injects
  request_input LangChain StructuredTool into self.tools. Called from
  both build_agent() and run_agent(). No-op when not running under A2A.
- Run ruff format on all A2A source and test files (17 reformatted)
- Alembic migration deferred: repo has pre-existing multiple heads
  that must be resolved first. SQLite auto-creates from SQLModel in
  tests. Migration will be created at PR merge time.

141 tests, all passing.
This commit is contained in:
Jordan Frazier
2026-03-27 21:05:58 -04:00
parent df244389ce
commit df68b1b372
19 changed files with 237 additions and 252 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/backend/base/langflow/api/a2a/agent_card.py
View File

@ -42,11 +42,7 @@ def generate_agent_card(flow: Flow, base_url: str) -> dict:
agent_name = getattr(flow, "a2a_name", None) or flow.name
# Resolve description: a2a_description > flow.description > default
agent_description = (
getattr(flow, "a2a_description", None)
or flow.description
or _DEFAULT_DESCRIPTION
)
agent_description = getattr(flow, "a2a_description", None) or flow.description or _DEFAULT_DESCRIPTION
# Resolve tags
tags = getattr(flow, "tags", None) or []

42
src/backend/base/langflow/api/a2a/config.py
View File

@ -10,26 +10,28 @@ import re
# Component types that indicate conversational / agentic capability.
# A flow must contain at least one of these to be eligible for A2A exposure.
_ELIGIBLE_COMPONENT_TYPES = frozenset({
# Agent components (primary use case)
"Agent",
"AgentComponent",
# LLM / Chat model components
"OpenAIModel",
"ChatOpenAI",
"AnthropicModel",
"ChatAnthropic",
"GoogleGenerativeAIModel",
"AzureChatOpenAI",
"OllamaModel",
"ChatOllama",
"GroqModel",
"HuggingFaceModel",
# Generic markers
"LLMModel",
"ChatModel",
"LanguageModel",
})
_ELIGIBLE_COMPONENT_TYPES = frozenset(
{
# Agent components (primary use case)
"Agent",
"AgentComponent",
# LLM / Chat model components
"OpenAIModel",
"ChatOpenAI",
"AnthropicModel",
"ChatAnthropic",
"GoogleGenerativeAIModel",
"AzureChatOpenAI",
"OllamaModel",
"ChatOllama",
"GroqModel",
"HuggingFaceModel",
# Generic markers
"LLMModel",
"ChatModel",
"LanguageModel",
}
)
# Slug format: lowercase alphanumeric + hyphens, 3-64 chars,
# must start/end with alphanumeric, no consecutive hyphens.

40
src/backend/base/langflow/api/a2a/flow_adapter.py
View File

@ -129,30 +129,38 @@ def _result_to_artifact(result, index: int = 0) -> dict | None:
if isinstance(results, dict):
message = results.get("message")
if isinstance(message, dict) and "text" in message:
parts.append({
"kind": "text",
"text": message["text"],
})
parts.append(
{
"kind": "text",
"text": message["text"],
}
)
elif isinstance(message, str):
parts.append({
"kind": "text",
"text": message,
})
parts.append(
{
"kind": "text",
"text": message,
}
)
# Check for structured data output
data = results.get("data")
if isinstance(data, dict):
parts.append({
"kind": "data",
"data": data,
})
parts.append(
{
"kind": "data",
"data": data,
}
)
if not parts:
# Fallback: serialize the entire result as data
parts.append({
"kind": "data",
"data": results if isinstance(results, dict) else {"result": str(results)},
})
parts.append(
{
"kind": "data",
"data": results if isinstance(results, dict) else {"result": str(results)},
}
)
return {
"artifactId": f"artifact-{index}",

11
src/backend/base/langflow/api/a2a/router.py
View File

@ -177,9 +177,7 @@ async def message_send(
if requested_task_id and _task_manager.has_pending_input(requested_task_id):
# Extract text from the follow-up message
parts = message.get("parts", [])
follow_up_text = " ".join(
p.get("text", "") for p in parts if p.get("kind") == "text"
)
follow_up_text = " ".join(p.get("text", "") for p in parts if p.get("kind") == "text")
await _task_manager.resolve_input(requested_task_id, follow_up_text)
task = await _task_manager.get_task(requested_task_id)
return task
@ -347,9 +345,7 @@ async def message_stream(
"task_manager": _task_manager,
"stream_bridge": bridge,
}
result = await _execute_flow(
flow, flow_inputs, session, a2a_context=a2a_context
)
result = await _execute_flow(flow, flow_inputs, session, a2a_context=a2a_context)
# Check if the task entered INPUT_REQUIRED during execution
# (this would be set by the request_input tool handler)
@ -539,8 +535,7 @@ async def update_a2a_config(
if not validate_flow_eligible_for_a2a(flow.data):
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail="This flow is not eligible for A2A exposure. "
"It must contain an Agent or LLM component.",
detail="This flow is not eligible for A2A exposure. It must contain an Agent or LLM component.",
)
# Apply updates

1
src/backend/base/pyproject.toml
View File

@ -83,6 +83,7 @@ dependencies = [
"networkx>=3.4.2,<4.0.0",
"json-repair>=0.30.3,<1.0.0",
"mcp>=1.17.0,<2.0.0",
"a2a-sdk>=0.3.0,<1.0.0",
"aiosqlite>=0.20.0,<1.0.0",
"greenlet>=3.1.1,<4.0.0",
"jsonquerylang>=1.1.1,<2.0.0",

24
src/backend/tests/unit/api/a2a/test_discovery.py
View File

@ -200,9 +200,7 @@ class TestA2AConfigManagement:
"""
flow = await _create_flow_via_api(client, logged_in_headers)
response = await _enable_a2a_on_flow(
client, logged_in_headers, flow["id"], slug="config-test"
)
response = await _enable_a2a_on_flow(client, logged_in_headers, flow["id"], slug="config-test")
assert response.status_code == 200
# Verify the config was persisted
@ -263,23 +261,15 @@ class TestA2AConfigManagement:
The slug is used in the public URL, so collisions would cause
routing ambiguity.
"""
flow1 = await _create_flow_via_api(
client, logged_in_headers, name="Flow One"
)
flow2 = await _create_flow_via_api(
client, logged_in_headers, name="Flow Two"
)
flow1 = await _create_flow_via_api(client, logged_in_headers, name="Flow One")
flow2 = await _create_flow_via_api(client, logged_in_headers, name="Flow Two")
# Enable first flow with slug
response1 = await _enable_a2a_on_flow(
client, logged_in_headers, flow1["id"], slug="unique-slug"
)
response1 = await _enable_a2a_on_flow(client, logged_in_headers, flow1["id"], slug="unique-slug")
assert response1.status_code == 200
# Try same slug on second flow → should fail
response2 = await _enable_a2a_on_flow(
client, logged_in_headers, flow2["id"], slug="unique-slug"
)
response2 = await _enable_a2a_on_flow(client, logged_in_headers, flow2["id"], slug="unique-slug")
assert response2.status_code == 409 # Conflict
async def test_invalid_slug_rejected(self, client: AsyncClient, logged_in_headers):
@ -306,9 +296,7 @@ class TestA2AConfigManagement:
data=_make_non_agent_flow_data(),
)
response = await _enable_a2a_on_flow(
client, logged_in_headers, flow["id"], slug="pipeline"
)
response = await _enable_a2a_on_flow(client, logged_in_headers, flow["id"], slug="pipeline")
assert response.status_code == 422 # Not eligible

26
src/backend/tests/unit/api/a2a/test_flow_adapter.py
View File

@ -181,9 +181,7 @@ class TestTranslateOutbound:
This is the most common case — the agent produces a text response.
"""
run_outputs = [
_make_run_output(message_text="Here is your analysis of the codebase.")
]
run_outputs = [_make_run_output(message_text="Here is your analysis of the codebase.")]
artifacts = await translate_outbound(run_outputs)
assert len(artifacts) == 1
@ -192,9 +190,7 @@ class TestTranslateOutbound:
async def test_structured_result_becomes_data_artifact(self):
"""When the flow returns structured data (dict), it becomes a data Part."""
run_outputs = [
_make_run_output(result_data={"score": 0.95, "labels": ["positive"]})
]
run_outputs = [_make_run_output(result_data={"score": 0.95, "labels": ["positive"]})]
artifacts = await translate_outbound(run_outputs)
assert len(artifacts) == 1
@ -242,14 +238,18 @@ def _make_run_output(
"""
outputs = []
if message_text is not None:
outputs.append({
"results": {"message": {"text": message_text}},
"messages": [{"message": message_text}],
})
outputs.append(
{
"results": {"message": {"text": message_text}},
"messages": [{"message": message_text}],
}
)
if result_data is not None:
outputs.append({
"results": {"data": result_data},
})
outputs.append(
{
"results": {"data": result_data},
}
)
return {
"inputs": {},
"outputs": outputs,

15
src/backend/tests/unit/api/a2a/test_input_required.py
View File

@ -72,6 +72,7 @@ def _make_a2a_message(text: str, context_id: str | None = None, task_id: str | N
def _mock_run_response(text: str = "Final answer"):
from unittest.mock import MagicMock
run_output = MagicMock()
run_output.inputs = {}
result_data = MagicMock()
@ -101,9 +102,7 @@ class TestInputRequiredFollowUp:
"""
@patch("langflow.api.v1.endpoints.simple_run_flow", new_callable=AsyncMock)
async def test_follow_up_to_input_required_task(
self, mock_run, client: AsyncClient, logged_in_headers
):
async def test_follow_up_to_input_required_task(self, mock_run, client: AsyncClient, logged_in_headers):
"""Send initial message, simulate INPUT_REQUIRED, send follow-up,
verify the task completes.
@ -154,9 +153,7 @@ class TestInputRequiredFollowUp:
assert body["id"] == task_id
@patch("langflow.api.v1.endpoints.simple_run_flow", new_callable=AsyncMock)
async def test_get_input_required_task_shows_question(
self, mock_run, client: AsyncClient, logged_in_headers
):
async def test_get_input_required_task_shows_question(self, mock_run, client: AsyncClient, logged_in_headers):
"""Polling an INPUT_REQUIRED task shows the agent's question.
The client uses GET /tasks/{id} to see what the agent is asking.
@ -211,7 +208,8 @@ class TestRequestInputTaskManagerIntegration:
await tm.update_state(task["id"], "working")
tool_info = create_request_input_tool(
task_id=task["id"], task_manager=tm,
task_id=task["id"],
task_manager=tm,
)
# Start the handler (will suspend and register pending input)
@ -235,7 +233,8 @@ class TestRequestInputTaskManagerIntegration:
await tm.update_state(task["id"], "working")
tool_info = create_request_input_tool(
task_id=task["id"], task_manager=tm,
task_id=task["id"],
task_manager=tm,
)
handler = tool_info["handler"]
future = asyncio.create_task(handler(question="Pick one"))

34
src/backend/tests/unit/api/a2a/test_input_required_e2e.py
View File

@ -95,9 +95,7 @@ class TestInputRequiredE2E:
3. Verifies the final response includes the resolved answer
"""
async def test_input_required_resolved_via_follow_up(
self, client: AsyncClient, logged_in_headers
):
async def test_input_required_resolved_via_follow_up(self, client: AsyncClient, logged_in_headers):
"""Full round-trip: send → INPUT_REQUIRED → follow-up → COMPLETED.
This tests:
@ -113,9 +111,16 @@ class TestInputRequiredE2E:
tool_event = asyncio.Event()
tool_response_holder: dict = {}
async def mock_simple_run_flow(*, flow=None, input_request=None, stream=False,
api_key_user=None, event_manager=None,
context=None, run_id=None):
async def mock_simple_run_flow(
*,
flow=None,
input_request=None,
stream=False,
api_key_user=None,
event_manager=None,
context=None,
run_id=None,
):
"""Simulate an agent that calls request_input mid-execution.
1. Signals INPUT_REQUIRED via task_manager
@ -208,9 +213,7 @@ class TestInputRequiredE2E:
artifact_text = body["artifacts"][0]["parts"][0]["text"]
assert "prod-us" in artifact_text
async def test_poll_shows_input_required_with_question(
self, client: AsyncClient, logged_in_headers
):
async def test_poll_shows_input_required_with_question(self, client: AsyncClient, logged_in_headers):
"""GET /tasks/{id} shows INPUT_REQUIRED state with the agent's question.
This is how a non-streaming client discovers that the agent
@ -223,9 +226,16 @@ class TestInputRequiredE2E:
tool_event = asyncio.Event()
tool_response_holder: dict = {}
async def mock_simple_run_flow(*, flow=None, input_request=None, stream=False,
api_key_user=None, event_manager=None,
context=None, run_id=None):
async def mock_simple_run_flow(
*,
flow=None,
input_request=None,
stream=False,
api_key_user=None,
event_manager=None,
context=None,
run_id=None,
):
all_tasks = await _task_manager.list_tasks()
current_task = all_tasks[-1]
await _task_manager.request_input(

40
src/backend/tests/unit/api/a2a/test_message_send.py
View File

@ -111,9 +111,7 @@ class TestMessageSendHappyPath:
"""
@patch("langflow.api.v1.endpoints.simple_run_flow", new_callable=AsyncMock)
async def test_send_text_message_returns_completed_task(
self, mock_run, client: AsyncClient, logged_in_headers
):
async def test_send_text_message_returns_completed_task(self, mock_run, client: AsyncClient, logged_in_headers):
"""Sending a text message executes the flow and returns a COMPLETED task.
This is the core happy path:
@ -145,9 +143,7 @@ class TestMessageSendHappyPath:
assert "Hello from the agent!" in first_artifact["parts"][0]["text"]
@patch("langflow.api.v1.endpoints.simple_run_flow", new_callable=AsyncMock)
async def test_task_has_context_id(
self, mock_run, client: AsyncClient, logged_in_headers
):
async def test_task_has_context_id(self, mock_run, client: AsyncClient, logged_in_headers):
"""The returned task preserves the contextId from the request.
This is essential for multi-turn conversations — the client
@ -168,9 +164,7 @@ class TestMessageSendHappyPath:
assert body["contextId"] == "ctx-my-conversation"
@patch("langflow.api.v1.endpoints.simple_run_flow", new_callable=AsyncMock)
async def test_idempotent_retry_returns_cached_result(
self, mock_run, client: AsyncClient, logged_in_headers
):
async def test_idempotent_retry_returns_cached_result(self, mock_run, client: AsyncClient, logged_in_headers):
"""Sending the same taskId twice returns the cached result.
The flow should NOT execute a second time. This prevents
@ -211,9 +205,7 @@ class TestMessageSendHappyPath:
class TestMessageSendAuth:
"""Tests for authentication on the message:send endpoint."""
async def test_unauthenticated_request_rejected(
self, client: AsyncClient, logged_in_headers
):
async def test_unauthenticated_request_rejected(self, client: AsyncClient, logged_in_headers):
"""message:send requires authentication.
Unlike the public AgentCard, message:send handles actual
@ -238,9 +230,7 @@ class TestMessageSendAuth:
class TestMessageSendErrors:
"""Tests for error handling in message:send."""
async def test_disabled_flow_returns_404(
self, client: AsyncClient, logged_in_headers
):
async def test_disabled_flow_returns_404(self, client: AsyncClient, logged_in_headers):
"""Sending to a slug that exists but has A2A disabled returns 404."""
# Create flow but don't enable A2A
flow = FlowCreate(name="Disabled", data=_make_agent_flow_data())
@ -255,9 +245,7 @@ class TestMessageSendErrors:
assert response.status_code == 404
@patch("langflow.api.v1.endpoints.simple_run_flow", new_callable=AsyncMock)
async def test_flow_execution_error_returns_failed_task(
self, mock_run, client: AsyncClient, logged_in_headers
):
async def test_flow_execution_error_returns_failed_task(self, mock_run, client: AsyncClient, logged_in_headers):
"""When the flow execution raises an exception, the task becomes FAILED.
The error message is included in the task status so the caller
@ -287,9 +275,7 @@ class TestTaskEndpoints:
"""Tests for task polling and cancellation endpoints."""
@patch("langflow.api.v1.endpoints.simple_run_flow", new_callable=AsyncMock)
async def test_get_task_returns_current_state(
self, mock_run, client: AsyncClient, logged_in_headers
):
async def test_get_task_returns_current_state(self, mock_run, client: AsyncClient, logged_in_headers):
"""GET /tasks/{task_id} returns the current task state.
This is the polling endpoint for clients that lost their
@ -316,9 +302,7 @@ class TestTaskEndpoints:
assert get_response.json()["id"] == task_id
assert get_response.json()["status"]["state"] == "completed"
async def test_get_nonexistent_task_returns_404(
self, client: AsyncClient, logged_in_headers
):
async def test_get_nonexistent_task_returns_404(self, client: AsyncClient, logged_in_headers):
"""Polling a task that doesn't exist returns 404."""
await _create_a2a_enabled_flow(client, logged_in_headers, slug="poll-agent-2")
@ -329,9 +313,7 @@ class TestTaskEndpoints:
assert response.status_code == 404
@patch("langflow.api.v1.endpoints.simple_run_flow", new_callable=AsyncMock)
async def test_cancel_task(
self, mock_run, client: AsyncClient, logged_in_headers
):
async def test_cancel_task(self, mock_run, client: AsyncClient, logged_in_headers):
"""POST /tasks/{task_id}:cancel sets the task to CANCELED.
Cancellation is best-effort — the flow may have already completed.
@ -357,9 +339,7 @@ class TestTaskEndpoints:
assert cancel_response.json()["status"]["state"] == "canceled"
@patch("langflow.api.v1.endpoints.simple_run_flow", new_callable=AsyncMock)
async def test_list_tasks_by_context(
self, mock_run, client: AsyncClient, logged_in_headers
):
async def test_list_tasks_by_context(self, mock_run, client: AsyncClient, logged_in_headers):
"""GET /tasks?contextId=... returns tasks for that conversation."""
mock_run.return_value = _mock_run_response("Response")
await _create_a2a_enabled_flow(client, logged_in_headers, slug="list-agent")

32
src/backend/tests/unit/api/a2a/test_message_stream.py
View File

@ -41,9 +41,7 @@ def _make_agent_flow_data() -> dict:
}
async def _create_a2a_enabled_flow(
client: AsyncClient, headers: dict, slug: str
) -> dict:
async def _create_a2a_enabled_flow(client: AsyncClient, headers: dict, slug: str) -> dict:
flow = FlowCreate(name=f"Flow {slug}", description="Test", data=_make_agent_flow_data())
response = await client.post("api/v1/flows/", json=flow.model_dump(), headers=headers)
assert response.status_code == 201
@ -112,9 +110,7 @@ class TestMessageStreamHappyPath:
"""Tests for successful streaming flow execution."""
@patch("langflow.api.v1.endpoints.simple_run_flow", new_callable=AsyncMock)
async def test_stream_returns_sse_response(
self, mock_run, client: AsyncClient, logged_in_headers
):
async def test_stream_returns_sse_response(self, mock_run, client: AsyncClient, logged_in_headers):
"""message:stream returns a streaming response with SSE media type.
The response should be text/event-stream, not application/json.
@ -132,9 +128,7 @@ class TestMessageStreamHappyPath:
assert "text/event-stream" in response.headers.get("content-type", "")
@patch("langflow.api.v1.endpoints.simple_run_flow", new_callable=AsyncMock)
async def test_stream_contains_completed_event(
self, mock_run, client: AsyncClient, logged_in_headers
):
async def test_stream_contains_completed_event(self, mock_run, client: AsyncClient, logged_in_headers):
"""The stream must end with a COMPLETED status event.
This tells the client the task is done and no more events
@ -155,9 +149,7 @@ class TestMessageStreamHappyPath:
assert any(e["status"]["state"] == "completed" for e in status_events)
@patch("langflow.api.v1.endpoints.simple_run_flow", new_callable=AsyncMock)
async def test_stream_contains_working_event(
self, mock_run, client: AsyncClient, logged_in_headers
):
async def test_stream_contains_working_event(self, mock_run, client: AsyncClient, logged_in_headers):
"""The stream should include a WORKING status event at the start.
This tells the client the flow has started executing.
@ -176,9 +168,7 @@ class TestMessageStreamHappyPath:
assert any(e["status"]["state"] == "working" for e in status_events)
@patch("langflow.api.v1.endpoints.simple_run_flow", new_callable=AsyncMock)
async def test_stream_preserves_task_and_context_ids(
self, mock_run, client: AsyncClient, logged_in_headers
):
async def test_stream_preserves_task_and_context_ids(self, mock_run, client: AsyncClient, logged_in_headers):
"""Every SSE event carries the taskId and contextId.
This lets clients correlate events with the right task
@ -208,9 +198,7 @@ class TestMessageStreamErrors:
"""Tests for error handling during streaming."""
@patch("langflow.api.v1.endpoints.simple_run_flow", new_callable=AsyncMock)
async def test_stream_error_produces_failed_event(
self, mock_run, client: AsyncClient, logged_in_headers
):
async def test_stream_error_produces_failed_event(self, mock_run, client: AsyncClient, logged_in_headers):
"""When flow execution fails, the stream includes a FAILED event.
The client should see the error without the connection just dying.
@ -229,9 +217,7 @@ class TestMessageStreamErrors:
status_events = [e for e in events if e.get("kind") == "status-update"]
assert any(e["status"]["state"] == "failed" for e in status_events)
async def test_stream_disabled_flow_returns_404(
self, client: AsyncClient, logged_in_headers
):
async def test_stream_disabled_flow_returns_404(self, client: AsyncClient, logged_in_headers):
"""Streaming to a non-existent slug returns 404."""
response = await client.post(
"/a2a/nonexistent/v1/message:stream",
@ -240,9 +226,7 @@ class TestMessageStreamErrors:
)
assert response.status_code == 404
async def test_stream_requires_auth(
self, client: AsyncClient, logged_in_headers
):
async def test_stream_requires_auth(self, client: AsyncClient, logged_in_headers):
"""message:stream requires authentication."""
await _create_a2a_enabled_flow(client, logged_in_headers, slug="auth-stream")

32
src/backend/tests/unit/api/a2a/test_multi_turn.py
View File

@ -44,9 +44,7 @@ def _make_agent_flow_data() -> dict:
}
async def _create_a2a_enabled_flow(
client: AsyncClient, headers: dict, slug: str
) -> dict:
async def _create_a2a_enabled_flow(client: AsyncClient, headers: dict, slug: str) -> dict:
flow = FlowCreate(name=f"Flow {slug}", description="Test", data=_make_agent_flow_data())
response = await client.post("api/v1/flows/", json=flow.model_dump(), headers=headers)
assert response.status_code == 201
@ -103,9 +101,7 @@ class TestSessionContinuity:
"""
@patch("langflow.api.v1.endpoints.simple_run_flow", new_callable=AsyncMock)
async def test_same_context_id_uses_same_session(
self, mock_run, client: AsyncClient, logged_in_headers
):
async def test_same_context_id_uses_same_session(self, mock_run, client: AsyncClient, logged_in_headers):
"""Two messages with the same contextId pass the same session_id
to simple_run_flow.
@ -143,9 +139,7 @@ class TestSessionContinuity:
assert session1.session_id.startswith("a2a-")
@patch("langflow.api.v1.endpoints.simple_run_flow", new_callable=AsyncMock)
async def test_different_context_ids_use_different_sessions(
self, mock_run, client: AsyncClient, logged_in_headers
):
async def test_different_context_ids_use_different_sessions(self, mock_run, client: AsyncClient, logged_in_headers):
"""Two messages with different contextIds use different session_ids.
This ensures conversation isolation — one caller's context
@ -193,9 +187,7 @@ class TestTaskLinkage:
"""
@patch("langflow.api.v1.endpoints.simple_run_flow", new_callable=AsyncMock)
async def test_each_turn_creates_separate_task(
self, mock_run, client: AsyncClient, logged_in_headers
):
async def test_each_turn_creates_separate_task(self, mock_run, client: AsyncClient, logged_in_headers):
"""Each message in a conversation produces a new Task.
Tasks are the unit of work in A2A — each has its own lifecycle
@ -219,9 +211,7 @@ class TestTaskLinkage:
assert len(set(task_ids)) == 3
@patch("langflow.api.v1.endpoints.simple_run_flow", new_callable=AsyncMock)
async def test_list_tasks_returns_all_turns(
self, mock_run, client: AsyncClient, logged_in_headers
):
async def test_list_tasks_returns_all_turns(self, mock_run, client: AsyncClient, logged_in_headers):
"""Listing tasks by contextId returns all turns in the conversation.
This is how a client retrieves the full conversation history
@ -256,9 +246,7 @@ class TestTaskLinkage:
assert len(tasks_b) == 1
@patch("langflow.api.v1.endpoints.simple_run_flow", new_callable=AsyncMock)
async def test_all_tasks_share_context_id(
self, mock_run, client: AsyncClient, logged_in_headers
):
async def test_all_tasks_share_context_id(self, mock_run, client: AsyncClient, logged_in_headers):
"""Every task in a conversation carries the same contextId.
The contextId is the conversation thread identifier — it must
@ -280,9 +268,7 @@ class TestTaskLinkage:
assert task["contextId"] == "ctx-preserved"
@patch("langflow.api.v1.endpoints.simple_run_flow", new_callable=AsyncMock)
async def test_all_turns_complete_independently(
self, mock_run, client: AsyncClient, logged_in_headers
):
async def test_all_turns_complete_independently(self, mock_run, client: AsyncClient, logged_in_headers):
"""Each turn completes independently — Turn 2 doesn't wait for Turn 1.
In v1, turns are independent flow executions. The only continuity
@ -314,9 +300,7 @@ class TestSingleTurn:
"""
@patch("langflow.api.v1.endpoints.simple_run_flow", new_callable=AsyncMock)
async def test_no_context_id_generates_unique_sessions(
self, mock_run, client: AsyncClient, logged_in_headers
):
async def test_no_context_id_generates_unique_sessions(self, mock_run, client: AsyncClient, logged_in_headers):
"""Two messages without contextId get different session_ids.
Without an explicit contextId, each message is a standalone

18
src/backend/tests/unit/api/a2a/test_request_input.py
View File

@ -41,9 +41,7 @@ def task_manager():
@pytest.fixture
async def task_with_tool(task_manager):
"""Create a task and its associated request_input tool."""
task = await task_manager.create_task(
flow_id="flow-1", context_id="ctx-1", task_id="task-input-1"
)
task = await task_manager.create_task(flow_id="flow-1", context_id="ctx-1", task_id="task-input-1")
tool_info = create_request_input_tool(
task_id=task["id"],
task_manager=task_manager,
@ -183,9 +181,7 @@ class TestTimeout:
The Agent can then decide to proceed without the info or fail.
"""
task = await task_manager.create_task(
flow_id="f", context_id="c", task_id="timeout-task"
)
task = await task_manager.create_task(flow_id="f", context_id="c", task_id="timeout-task")
tool_info = create_request_input_tool(
task_id=task["id"],
task_manager=task_manager,
@ -216,13 +212,12 @@ class TestMultipleRounds:
Round 1: "Which env?" → "staging"
Round 2: "Which region?" → "us-west-2"
"""
task = await task_manager.create_task(
flow_id="f", context_id="c", task_id="multi-q"
)
task = await task_manager.create_task(flow_id="f", context_id="c", task_id="multi-q")
# Round 1
tool_info_1 = create_request_input_tool(
task_id=task["id"], task_manager=task_manager,
task_id=task["id"],
task_manager=task_manager,
)
handler_1 = tool_info_1["handler"]
future_1 = asyncio.create_task(handler_1(question="Which env?"))
@ -234,7 +229,8 @@ class TestMultipleRounds:
# Round 2 — new tool instance (agent creates a new call)
tool_info_2 = create_request_input_tool(
task_id=task["id"], task_manager=task_manager,
task_id=task["id"],
task_manager=task_manager,
)
handler_2 = tool_info_2["handler"]
future_2 = asyncio.create_task(handler_2(question="Which region?"))

8
src/backend/tests/unit/api/a2a/test_smoke.py
View File

@ -79,9 +79,7 @@ class TestA2ASmoke:
description="An agent for the A2A smoke test",
data=_make_agent_flow_data(),
)
create_resp = await client.post(
"api/v1/flows/", json=flow.model_dump(), headers=logged_in_headers
)
create_resp = await client.post("api/v1/flows/", json=flow.model_dump(), headers=logged_in_headers)
assert create_resp.status_code == 201
flow_id = create_resp.json()["id"]
@ -104,9 +102,7 @@ class TestA2ASmoke:
assert config["a2a_agent_slug"] == "smoke-agent"
# Verify config readback
config_get = await client.get(
f"api/v1/flows/{flow_id}/a2a-config", headers=logged_in_headers
)
config_get = await client.get(f"api/v1/flows/{flow_id}/a2a-config", headers=logged_in_headers)
assert config_get.status_code == 200
assert config_get.json()["a2a_name"] == "Smoke Agent"

32
src/backend/tests/unit/api/a2a/test_streaming.py
View File

@ -87,9 +87,7 @@ class TestTokenStreaming:
bridge = A2AStreamBridge(task_id="task-1", context_id="ctx-1")
for token in ["Hello", " world", "!"]:
await bridge.process_langflow_event(
_make_langflow_event("token", {"chunk": token})
)
await bridge.process_langflow_event(_make_langflow_event("token", {"chunk": token}))
events = await _collect_a2a_events(bridge, 3)
assert len(events) == 3
@ -113,9 +111,7 @@ class TestStatusTransitions:
bridge = A2AStreamBridge(task_id="task-1", context_id="ctx-1")
end_data = {"result": {"outputs": [{"inputs": {}, "outputs": []}]}}
await bridge.process_langflow_event(
_make_langflow_event("end", end_data)
)
await bridge.process_langflow_event(_make_langflow_event("end", end_data))
events = await _collect_a2a_events(bridge, 1)
assert len(events) >= 1
@ -132,9 +128,7 @@ class TestStatusTransitions:
"""
bridge = A2AStreamBridge(task_id="task-1", context_id="ctx-1")
await bridge.process_langflow_event(
_make_langflow_event("error", {"error": "LLM rate limit exceeded"})
)
await bridge.process_langflow_event(_make_langflow_event("error", {"error": "LLM rate limit exceeded"}))
events = await _collect_a2a_events(bridge, 1)
assert len(events) == 1
@ -149,9 +143,7 @@ class TestStatusTransitions:
"""
bridge = A2AStreamBridge(task_id="task-1", context_id="ctx-1")
await bridge.process_langflow_event(
_make_langflow_event("end_vertex", {"build_data": "vertex-abc"})
)
await bridge.process_langflow_event(_make_langflow_event("end_vertex", {"build_data": "vertex-abc"}))
events = await _collect_a2a_events(bridge, 1)
assert len(events) == 1
@ -184,17 +176,11 @@ class TestStreamLifecycle:
bridge = A2AStreamBridge(task_id="task-1", context_id="ctx-1")
# Tokens
await bridge.process_langflow_event(
_make_langflow_event("token", {"chunk": "Hello"})
)
await bridge.process_langflow_event(
_make_langflow_event("token", {"chunk": " world"})
)
await bridge.process_langflow_event(_make_langflow_event("token", {"chunk": "Hello"}))
await bridge.process_langflow_event(_make_langflow_event("token", {"chunk": " world"}))
# End
await bridge.process_langflow_event(
_make_langflow_event("end", {"result": {}})
)
await bridge.process_langflow_event(_make_langflow_event("end", {"result": {}}))
events = await _collect_a2a_events(bridge, 10)
@ -213,9 +199,7 @@ class TestStreamLifecycle:
"""
bridge = A2AStreamBridge(task_id="task-1", context_id="ctx-1")
await bridge.process_langflow_event(
_make_langflow_event("build_start", {"vertex_id": "v1"})
)
await bridge.process_langflow_event(_make_langflow_event("build_start", {"vertex_id": "v1"}))
# Should produce no output events
events = await _collect_a2a_events(bridge, 1, timeout=0.1)

24
src/backend/tests/unit/api/a2a/test_task_cleanup.py
View File

@ -37,9 +37,7 @@ class TestTaskCleanup:
await task_manager.update_state(task["id"], "completed")
# Backdate the task
task_manager._tasks["old-done"]["_updated_at"] = (
datetime.now(timezone.utc) - timedelta(hours=25)
).isoformat()
task_manager._tasks["old-done"]["_updated_at"] = (datetime.now(timezone.utc) - timedelta(hours=25)).isoformat()
pruned = await task_manager.cleanup_expired_tasks(ttl_seconds=86400) # 24h
@ -52,9 +50,7 @@ class TestTaskCleanup:
await task_manager.update_state(task["id"], "working")
await task_manager.update_state(task["id"], "failed", error="boom")
task_manager._tasks["old-fail"]["_updated_at"] = (
datetime.now(timezone.utc) - timedelta(hours=25)
).isoformat()
task_manager._tasks["old-fail"]["_updated_at"] = (datetime.now(timezone.utc) - timedelta(hours=25)).isoformat()
pruned = await task_manager.cleanup_expired_tasks(ttl_seconds=86400)
assert pruned == 1
@ -97,9 +93,7 @@ class TestTaskCleanup:
task = await task_manager.create_task(flow_id="f", context_id="c", task_id="old-ir")
await task_manager.update_state(task["id"], "input-required")
task_manager._tasks["old-ir"]["_updated_at"] = (
datetime.now(timezone.utc) - timedelta(hours=48)
).isoformat()
task_manager._tasks["old-ir"]["_updated_at"] = (datetime.now(timezone.utc) - timedelta(hours=48)).isoformat()
pruned = await task_manager.cleanup_expired_tasks(ttl_seconds=86400)
assert pruned == 0
@ -113,9 +107,7 @@ class TestTaskCleanup:
# Old completed → prune
t1 = await task_manager.create_task(flow_id="f", context_id="c", task_id="t1")
await task_manager.update_state(t1["id"], "completed")
task_manager._tasks["t1"]["_updated_at"] = (
datetime.now(timezone.utc) - timedelta(hours=25)
).isoformat()
task_manager._tasks["t1"]["_updated_at"] = (datetime.now(timezone.utc) - timedelta(hours=25)).isoformat()
# New completed → keep
t2 = await task_manager.create_task(flow_id="f", context_id="c", task_id="t2")
@ -124,16 +116,12 @@ class TestTaskCleanup:
# Old working → keep (active)
t3 = await task_manager.create_task(flow_id="f", context_id="c", task_id="t3")
await task_manager.update_state(t3["id"], "working")
task_manager._tasks["t3"]["_updated_at"] = (
datetime.now(timezone.utc) - timedelta(hours=25)
).isoformat()
task_manager._tasks["t3"]["_updated_at"] = (datetime.now(timezone.utc) - timedelta(hours=25)).isoformat()
# Old canceled → prune
t4 = await task_manager.create_task(flow_id="f", context_id="c", task_id="t4")
await task_manager.update_state(t4["id"], "canceled")
task_manager._tasks["t4"]["_updated_at"] = (
datetime.now(timezone.utc) - timedelta(hours=25)
).isoformat()
task_manager._tasks["t4"]["_updated_at"] = (datetime.now(timezone.utc) - timedelta(hours=25)).isoformat()
pruned = await task_manager.cleanup_expired_tasks(ttl_seconds=86400)
assert pruned == 2 # t1 and t4

20
src/backend/tests/unit/api/a2a/test_task_manager.py
View File

@ -98,9 +98,7 @@ class TestStateTransitions:
await task_manager.update_state(task["id"], "working")
artifacts = [{"parts": [{"kind": "text", "text": "Result"}]}]
updated = await task_manager.update_state(
task["id"], "completed", artifacts=artifacts
)
updated = await task_manager.update_state(task["id"], "completed", artifacts=artifacts)
assert updated["status"]["state"] == "completed"
assert updated["artifacts"] == artifacts
@ -110,9 +108,7 @@ class TestStateTransitions:
task = await task_manager.create_task(flow_id="f", context_id="c")
await task_manager.update_state(task["id"], "working")
updated = await task_manager.update_state(
task["id"], "failed", error="Flow execution failed: division by zero"
)
updated = await task_manager.update_state(task["id"], "failed", error="Flow execution failed: division by zero")
assert updated["status"]["state"] == "failed"
assert "division by zero" in updated["status"]["message"]["parts"][0]["text"]
@ -192,9 +188,7 @@ class TestIdempotentRetry:
This prevents duplicate execution when a client retries after
a network timeout (the original request may have succeeded).
"""
task = await task_manager.create_task(
flow_id="f", context_id="c", task_id="retry-me"
)
task = await task_manager.create_task(flow_id="f", context_id="c", task_id="retry-me")
await task_manager.update_state(task["id"], "working")
artifacts = [{"parts": [{"kind": "text", "text": "Original result"}]}]
await task_manager.update_state(task["id"], "completed", artifacts=artifacts)
@ -211,9 +205,7 @@ class TestIdempotentRetry:
Failed tasks should be retried — the failure may have been
transient (network error, rate limit, etc.).
"""
task = await task_manager.create_task(
flow_id="f", context_id="c", task_id="retry-failed"
)
task = await task_manager.create_task(flow_id="f", context_id="c", task_id="retry-failed")
await task_manager.update_state(task["id"], "working")
await task_manager.update_state(task["id"], "failed", error="Timeout")
@ -226,9 +218,7 @@ class TestIdempotentRetry:
The client should see that the task is still in progress
rather than starting a duplicate execution.
"""
task = await task_manager.create_task(
flow_id="f", context_id="c", task_id="in-progress"
)
task = await task_manager.create_task(flow_id="f", context_id="c", task_id="in-progress")
await task_manager.update_state(task["id"], "working")
existing = await task_manager.handle_retry("in-progress")

63
src/lfx/src/lfx/base/agents/agent.py
View File

@ -143,10 +143,72 @@ class LCAgentComponent(Component):
return messages
def _inject_a2a_tools(self) -> None:
"""Inject the request_input tool when running under A2A execution.
Checks graph.context for an "a2a" key set by the A2A router.
If present, creates and appends a request_input tool that allows
the LLM to ask the calling agent for clarification mid-execution.
This method is called from build_agent() and run_agent() before
the AgentExecutor is created. It's a no-op when not running
under A2A (normal playground execution).
"""
# Check if we have a graph context with A2A execution info
if not hasattr(self, "graph") or self.graph is None:
return
a2a_ctx = self.graph.context.get("a2a") if hasattr(self.graph, "context") else None
if not a2a_ctx:
return
task_id = a2a_ctx.get("task_id")
task_manager = a2a_ctx.get("task_manager")
if not task_id or not task_manager:
return
try:
from langflow.api.a2a.request_input_tool import create_request_input_tool
tool_info = create_request_input_tool(
task_id=task_id,
task_manager=task_manager,
)
# Store the tool info on the context so the router can access
# the event and response_holder for follow-up resolution
a2a_ctx["request_input_tool_info"] = tool_info
# Create a LangChain StructuredTool from our tool info
from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field
class RequestInputSchema(BaseModel):
question: str = Field(description="The question to ask the calling agent")
lc_tool = StructuredTool(
name=tool_info["name"],
description=tool_info["description"],
coroutine=tool_info["handler"],
args_schema=RequestInputSchema,
)
if self.tools is None:
self.tools = []
self.tools.append(lc_tool)
logger.info(f"A2A: Injected request_input tool for task {task_id}")
except ImportError:
logger.debug("A2A request_input_tool not available — skipping injection")
except Exception:
logger.exception("Failed to inject A2A request_input tool")
async def run_agent(
self,
agent: Runnable | BaseSingleActionAgent | BaseMultiActionAgent | AgentExecutor,
) -> Message:
self._inject_a2a_tools()
if isinstance(agent, AgentExecutor):
runnable = agent
else:
@ -322,6 +384,7 @@ class LCToolsAgentComponent(LCAgentComponent):
def build_agent(self) -> AgentExecutor:
self.validate_tool_names()
self._inject_a2a_tools()
agent = self.create_agent_runnable()
return AgentExecutor.from_agent_and_tools(
agent=RunnableAgent(runnable=agent, input_keys_arg=["input"], return_keys_arg=["output"]),

21
uv.lock generated
View File

@ -30,6 +30,22 @@ overrides = [
{ name = "z3-solver", specifier = "<4.15.7" },
]
[[package]]
name = "a2a-sdk"
version = "0.3.25"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "google-api-core" },
{ name = "httpx" },
{ name = "httpx-sse" },
{ name = "protobuf" },
{ name = "pydantic" },
]
sdist = { url = "https://files.pythonhosted.org/packages/55/83/3c99b276d09656cce039464509f05bf385e5600d6dc046a131bbcf686930/a2a_sdk-0.3.25.tar.gz", hash = "sha256:afda85bab8d6af0c5d15e82f326c94190f6be8a901ce562d045a338b7127242f", size = 270638, upload-time = "2026-03-10T13:08:46.417Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/bd/f9/6a62520b7ecb945188a6e1192275f4732ff9341cd4629bc975a6c146aeab/a2a_sdk-0.3.25-py3-none-any.whl", hash = "sha256:2fce38faea82eb0b6f9f9c2bcf761b0d78612c80ef0e599b50d566db1b2654b5", size = 149609, upload-time = "2026-03-10T13:08:44.7Z" },
]
[[package]]
name = "accelerate"
version = "1.13.0"
@ -3948,6 +3964,7 @@ dependencies = [
{ name = "griffecli" },
{ name = "griffelib" },
]
sdist = { url = "https://files.pythonhosted.org/packages/04/56/28a0accac339c164b52a92c6cfc45a903acc0c174caa5c1713803467b533/griffe-2.0.0.tar.gz", hash = "sha256:c68979cd8395422083a51ea7cf02f9c119d889646d99b7b656ee43725de1b80f", size = 293906, upload-time = "2026-03-23T21:06:53.402Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/8b/94/ee21d41e7eb4f823b94603b9d40f86d3c7fde80eacc2c3c71845476dddaa/griffe-2.0.0-py3-none-any.whl", hash = "sha256:5418081135a391c3e6e757a7f3f156f1a1a746cc7b4023868ff7d5e2f9a980aa", size = 5214, upload-time = "2026-02-09T19:09:44.105Z" },
]
@ -3960,6 +3977,7 @@ dependencies = [
{ name = "colorama" },
{ name = "griffelib" },
]
sdist = { url = "https://files.pythonhosted.org/packages/a4/f8/2e129fd4a86e52e58eefe664de05e7d502decf766e7316cc9e70fdec3e18/griffecli-2.0.0.tar.gz", hash = "sha256:312fa5ebb4ce6afc786356e2d0ce85b06c1c20d45abc42d74f0cda65e159f6ef", size = 56213, upload-time = "2026-03-23T21:06:54.8Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/e6/ed/d93f7a447bbf7a935d8868e9617cbe1cadf9ee9ee6bd275d3040fbf93d60/griffecli-2.0.0-py3-none-any.whl", hash = "sha256:9f7cd9ee9b21d55e91689358978d2385ae65c22f307a63fb3269acf3f21e643d", size = 9345, upload-time = "2026-02-09T19:09:42.554Z" },
]
@ -3968,6 +3986,7 @@ wheels = [
name = "griffelib"
version = "2.0.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/ad/06/eccbd311c9e2b3ca45dbc063b93134c57a1ccc7607c5e545264ad092c4a9/griffelib-2.0.0.tar.gz", hash = "sha256:e504d637a089f5cab9b5daf18f7645970509bf4f53eda8d79ed71cce8bd97934", size = 166312, upload-time = "2026-03-23T21:06:55.954Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/4d/51/c936033e16d12b627ea334aaaaf42229c37620d0f15593456ab69ab48161/griffelib-2.0.0-py3-none-any.whl", hash = "sha256:01284878c966508b6d6f1dbff9b6fa607bc062d8261c5c7253cb285b06422a7f", size = 142004, upload-time = "2026-02-09T19:09:40.561Z" },
]
@ -5994,6 +6013,7 @@ name = "langflow-base"
version = "0.9.0"
source = { editable = "src/backend/base" }
dependencies = [
{ name = "a2a-sdk" },
{ name = "aiofile" },
{ name = "aiofiles" },
{ name = "aiosqlite" },
@ -6686,6 +6706,7 @@ dev = [
[package.metadata]
requires-dist = [
{ name = "a2a-sdk", specifier = ">=0.3.0,<1.0.0" },
{ name = "ag2", marker = "extra == 'ag2'", specifier = ">=0.1.0" },
{ name = "agent-lifecycle-toolkit", marker = "extra == 'altk'", specifier = "~=0.4.4" },
{ name = "aioboto3", marker = "extra == 'aioboto3'", specifier = ">=15.2.0,<16.0.0" },