diff --git a/docs/mkdocs/en/graph.md b/docs/mkdocs/en/graph.md
index 83ed3038..e93c6120 100644
--- a/docs/mkdocs/en/graph.md
+++ b/docs/mkdocs/en/graph.md
@@ -429,6 +429,10 @@ Common options:
 - input_mapper / output_mapper: Parent-child state mapping (explicit configuration recommended)
 - config / callbacks: Same as add_node
 
+When a child Agent emits a `LongRunningEvent`, `add_agent_node` promotes it to a parent GraphAgent `interrupt`: the parent graph does not execute downstream nodes, and the Runner event preserves the original tool name and arguments. After the client submits the matching `FunctionResponse`, the parent graph resumes the current Agent node and the SDK maps the response back to the child's original function call. The graph continues only after the child Agent reaches a final result. This supports multiple HITL rounds in one node and persists child state in the SessionService-backed checkpoint for process-restart recovery.
+
+When the Agent node is a TeamAgent, the Leader can use the same HITL flow. Regular Team Members still must not be configured with or invoke `LongRunningFunctionTool`.
+
 GraphAgent does not require (nor support) registering Agent nodes via sub_agents; composition relationships are handled uniformly through add_agent_node.
 
 ## Advanced Usage
diff --git a/docs/mkdocs/en/model.md b/docs/mkdocs/en/model.md
index 5b9618e7..539dd617 100644
--- a/docs/mkdocs/en/model.md
+++ b/docs/mkdocs/en/model.md
@@ -121,6 +121,32 @@ model = OpenAIModel(
 )
 ```
 
+#### Responses API
+
+`OpenAIModel` uses Chat Completions by default. Enable the Responses API explicitly for OpenAI or compatible
+providers that expose `/v1/responses`:
+
+```python
+model = OpenAIModel(
+    model_name="your-responses-model",
+    api_key="your-api-key",
+    base_url="https://api.openai.com/v1",
+    use_responses_api=True,
+    responses_api_params={
+        "store": False,
+        "reasoning": {"summary": "auto"},
+    },
+)
+```
+
+The switch is opt-in so existing OpenAI-compatible providers continue to use Chat Completions. The adapter maps
+conversation history, function calls and `function_call_output` items, structured output, semantic streaming events,
+reasoning summaries, and token usage into the existing tRPC-Agent types. When `store=False`, the SDK automatically
+requests `reasoning.encrypted_content` so reasoning items can be replayed with tool outputs in the next turn.
+
+`responses_api_params` accepts Responses-only fields such as `store`, `reasoning`, `include`, and `truncation`.
+`model`, `input`, and `stream` are managed by `OpenAIModel` and cannot be overridden there.
+
 #### Advanced Usage
 
 Since version `1.1.10`, `OpenAIModel` supports passing a shared HTTP client provider to enable connection reuse. By default, `OpenAIModel` creates a temporary HTTP client for each model-service request. If you want to reuse connections, use the following configuration:
diff --git a/docs/mkdocs/zh/graph.md b/docs/mkdocs/zh/graph.md
index f5683e0e..15df04ed 100644
--- a/docs/mkdocs/zh/graph.md
+++ b/docs/mkdocs/zh/graph.md
@@ -428,6 +428,10 @@ graph.add_agent_node(
 - input_mapper / output_mapper：父子状态映射（推荐显式配置）
 - config / callbacks：同 add_node
 
+当子 Agent 发出 `LongRunningEvent` 时，`add_agent_node` 会将其提升为父 GraphAgent 的 `interrupt`：父图不会执行后继节点，Runner 返回的事件保留原工具名和参数。客户端提交对应 `FunctionResponse` 后，父图恢复当前 Agent 节点，SDK 将响应映射回子 Agent 的原始 function call；只有子 Agent 最终完成后父图才继续。该机制支持同一节点多轮 HITL，并将 child state 写入 SessionService-backed checkpoint，服务重启后仍可恢复。
+
+TeamAgent 作为 Agent 节点时同样支持 Leader 发起 HITL；普通 Team Member 仍不允许配置或调用 `LongRunningFunctionTool`。
+
 GraphAgent 不需要（也不支持）通过 sub_agents 注册 Agent 节点；组合关系统一用 add_agent_node 完成。
 
 ## 进阶用法
diff --git a/docs/mkdocs/zh/model.md b/docs/mkdocs/zh/model.md
index 16e2320b..6209891b 100644
--- a/docs/mkdocs/zh/model.md
+++ b/docs/mkdocs/zh/model.md
@@ -121,6 +121,31 @@ model = OpenAIModel(
 )
 ```
 
+#### Responses API
+
+`OpenAIModel` 默认仍使用 Chat Completions。对于 OpenAI 或提供 `/v1/responses` 的兼容服务，需要显式开启：
+
+```python
+model = OpenAIModel(
+    model_name="your-responses-model",
+    api_key="your-api-key",
+    base_url="https://api.openai.com/v1",
+    use_responses_api=True,
+    responses_api_params={
+        "store": False,
+        "reasoning": {"summary": "auto"},
+    },
+)
+```
+
+该开关默认关闭，现有 OpenAI-compatible 服务不会改变调用路径。适配层会把多轮消息、函数调用及
+`function_call_output`、结构化输出、语义化流式事件、reasoning summary 和 token usage 映射为现有
+tRPC-Agent 类型。设置 `store=False` 时，SDK 会自动请求 `reasoning.encrypted_content`，以便下一轮
+连同工具结果一起回放 reasoning item。
+
+`responses_api_params` 可传入 `store`、`reasoning`、`include`、`truncation` 等 Responses 专用字段；
+`model`、`input`、`stream` 由 `OpenAIModel` 管理，不能在此覆盖。
+
 #### 高级用法
 
 从版本 `1.1.10`之后 OpenAIModel 支持传入共享的 http client 来解决连接复用的场景，当前的 OpenAIModel 默认每次都会创建临时的 http client 去访问模型服务；如果期望连接复用可以使用如下的方式
diff --git a/pyproject.toml b/pyproject.toml
index db0e7828..33c466a5 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -25,7 +25,7 @@ classifiers = [
 ]
 dependencies = [
     "pydantic>=2.11.3",
-    "openai>=1.3.0",
+    "openai>=1.66.0",
     "mcp>=1.10.1",
     "aiohttp",
     "httpx>=0.27.0",
diff --git a/tests/models/test_openai_responses_model.py b/tests/models/test_openai_responses_model.py
new file mode 100644
index 00000000..6b6b3dfc
--- /dev/null
+++ b/tests/models/test_openai_responses_model.py
@@ -0,0 +1,427 @@
+# Tencent is pleased to support the open source community by making tRPC-Agent-Python available.
+#
+# Copyright (C) 2026 Tencent. All rights reserved.
+#
+# tRPC-Agent-Python is licensed under Apache-2.0.
+
+from unittest.mock import AsyncMock, Mock, patch
+
+import pytest
+from trpc_agent_sdk.models import LlmRequest, OpenAIModel
+from trpc_agent_sdk.types import Content, GenerateContentConfig, Part
+
+
+def _model(**kwargs):
+    """Create an OpenAIModel with test defaults."""
+    kwargs.setdefault("model_name", "gpt-4")
+    kwargs.setdefault("api_key", "test_key")
+    return OpenAIModel(**kwargs)
+
+
+def _request(contents, config=None, streaming_tool_names=None):
+    """Create an LlmRequest for Responses API tests."""
+    request = LlmRequest(contents=contents, config=config, tools_dict={})
+    if streaming_tool_names is not None:
+        request.streaming_tool_names = streaming_tool_names
+    return request
+
+
+# ---------------------------------------------------------------------------
+# Responses API
+# ---------------------------------------------------------------------------
+
+
+class TestOpenAIResponsesAPI:
+    """Tests for the opt-in OpenAI Responses transport."""
+
+    def test_is_disabled_by_default_and_rejects_managed_overrides(self):
+        model = _model()
+        assert model.use_responses_api is False
+
+        with pytest.raises(ValueError, match="cannot override managed parameters: input"):
+            _model(use_responses_api=True, responses_api_params={"input": "override"})
+
+    def test_converts_tool_history_and_function_definitions(self):
+        model = _model(use_responses_api=True)
+        messages = [
+            {
+                "role": "assistant",
+                "content": "Checking",
+                "tool_calls": [
+                    {
+                        "id": "call_weather",
+                        "type": "function",
+                        "function": {
+                            "name": "weather",
+                            "arguments": '{"city":"Shenzhen"}',
+                        },
+                    }
+                ],
+            },
+            {"role": "tool", "tool_call_id": "call_weather", "content": '{"temperature":30}'},
+        ]
+
+        items = model._convert_messages_to_responses_input(messages)
+        assert items == [
+            {"role": "assistant", "content": "Checking"},
+            {
+                "type": "function_call",
+                "call_id": "call_weather",
+                "name": "weather",
+                "arguments": '{"city":"Shenzhen"}',
+            },
+            {
+                "type": "function_call_output",
+                "call_id": "call_weather",
+                "output": '{"temperature":30}',
+            },
+        ]
+        tools = model._convert_tools_to_responses_format(
+            [
+                {
+                    "type": "function",
+                    "function": {
+                        "name": "weather",
+                        "description": "Get weather",
+                        "parameters": {"type": "object", "properties": {}},
+                    },
+                }
+            ]
+        )
+        assert tools == [
+            {
+                "type": "function",
+                "name": "weather",
+                "description": "Get weather",
+                "parameters": {"type": "object", "properties": {}},
+            }
+        ]
+
+    def test_converts_multimodal_input(self):
+        model = _model(use_responses_api=True)
+
+        items = model._convert_messages_to_responses_input(
+            [
+                {
+                    "role": "user",
+                    "content": [
+                        {
+                            "type": "text",
+                            "text": "Inspect this image",
+                        },
+                        {
+                            "type": "image_url",
+                            "image_url": {
+                                "url": "data:image/png;base64,AAAA",
+                                "detail": "high",
+                            },
+                        },
+                    ],
+                }
+            ]
+        )
+
+        assert items == [
+            {
+                "role": "user",
+                "content": [
+                    {
+                        "type": "input_text",
+                        "text": "Inspect this image",
+                    },
+                    {
+                        "type": "input_image",
+                        "image_url": "data:image/png;base64,AAAA",
+                        "detail": "high",
+                    },
+                ],
+            }
+        ]
+
+    def test_preserves_empty_reasoning_item_for_tool_continuation(self):
+        model = _model(use_responses_api=True)
+        raw_reasoning = {
+            "id": "rs_empty",
+            "type": "reasoning",
+            "encrypted_content": "encrypted-reasoning",
+            "summary": [],
+        }
+
+        response = model._create_responses_response(
+            {
+                "id": "resp_tool",
+                "status": "completed",
+                "output": [
+                    raw_reasoning,
+                    {
+                        "type": "function_call",
+                        "call_id": "call_1",
+                        "name": "lookup",
+                        "arguments": "{}",
+                    },
+                ],
+            }
+        )
+
+        thought = response.content.parts[0]
+        assert thought.thought is True
+        assert thought.text == ""
+        replay_request = _request([Content(parts=response.content.parts, role="model")])
+        replay_items = model._convert_messages_to_responses_input(model._format_messages(replay_request))
+        assert replay_items[0] == raw_reasoning
+        assert replay_items[1]["type"] == "function_call"
+
+    def test_converts_structured_output_and_failed_response(self):
+        model = _model(use_responses_api=True)
+        text = model._convert_response_format_to_responses(
+            {
+                "type": "json_schema",
+                "json_schema": {
+                    "name": "answer",
+                    "schema": {"type": "object", "properties": {}},
+                    "strict": True,
+                },
+            }
+        )
+        assert text == {
+            "format": {
+                "type": "json_schema",
+                "name": "answer",
+                "schema": {"type": "object", "properties": {}},
+                "strict": True,
+            },
+        }
+
+        response = model._create_responses_response(
+            {
+                "id": "resp_failed",
+                "status": "failed",
+                "error": {"code": "server_error", "message": "upstream failed"},
+                "output": [],
+            }
+        )
+        assert response.error_code == "server_error"
+        assert response.error_message == "upstream failed"
+
+    @pytest.mark.asyncio
+    async def test_non_streaming_uses_responses_create_and_maps_output(self):
+        model = _model(
+            use_responses_api=True,
+            responses_api_params={"store": False, "truncation": "auto"},
+        )
+        request = _request(
+            [Content(parts=[Part.from_text(text="Hello")], role="user")],
+            GenerateContentConfig(max_output_tokens=128),
+        )
+        response = Mock()
+        response.model_dump.return_value = {
+            "id": "resp_123",
+            "status": "completed",
+            "output": [
+                {
+                    "id": "rs_123",
+                    "type": "reasoning",
+                    "encrypted_content": "encrypted-reasoning",
+                    "summary": [{"type": "summary_text", "text": "Check context"}],
+                },
+                {
+                    "type": "message",
+                    "role": "assistant",
+                    "content": [{"type": "output_text", "text": "Hello back"}],
+                },
+                {
+                    "type": "function_call",
+                    "call_id": "call_123",
+                    "name": "lookup",
+                    "arguments": '{"query":"hello"}',
+                },
+            ],
+            "usage": {
+                "input_tokens": 10,
+                "output_tokens": 7,
+                "total_tokens": 17,
+                "input_tokens_details": {"cached_tokens": 4},
+                "output_tokens_details": {"reasoning_tokens": 2},
+            },
+        }
+        captured = {}
+
+        async def create(**kwargs):
+            captured.update(kwargs)
+            return response
+
+        with patch.object(model, "_create_async_client") as client_factory:
+            client = AsyncMock()
+            client.responses.create = create
+            client_factory.return_value = client
+            responses = [item async for item in model.generate_async(request, stream=False)]
+
+        assert captured["model"] == "gpt-4"
+        assert captured["input"] == [{"role": "user", "content": "Hello"}]
+        assert captured["max_output_tokens"] == 128
+        assert captured["store"] is False
+        assert captured["truncation"] == "auto"
+        assert captured["include"] == ["reasoning.encrypted_content"]
+        assert "messages" not in captured
+        assert "max_completion_tokens" not in captured
+        result = responses[0]
+        assert result.response_id == "resp_123"
+        assert [part.text for part in result.content.parts if part.text] == ["Check context", "Hello back"]
+        assert result.content.parts[0].thought is True
+        assert result.content.parts[-1].function_call.id == "call_123"
+        assert result.usage_metadata.prompt_token_count == 10
+        assert result.usage_metadata.candidates_token_count == 7
+        assert result.usage_metadata.thoughts_token_count == 2
+        assert result.usage_metadata.cache_read_input_tokens == 4
+
+        replay_request = _request([Content(parts=result.content.parts, role="model")])
+        replay_items = model._convert_messages_to_responses_input(model._format_messages(replay_request))
+        assert replay_items[0] == {
+            "id": "rs_123",
+            "type": "reasoning",
+            "encrypted_content": "encrypted-reasoning",
+            "summary": [{"type": "summary_text", "text": "Check context"}],
+        }
+        assert replay_items[1] == {"role": "assistant", "content": "Hello back"}
+        assert replay_items[2]["type"] == "function_call"
+
+    @pytest.mark.asyncio
+    async def test_streaming_maps_text_reasoning_tool_calls_and_usage(self):
+        model = _model(use_responses_api=True)
+        request = _request(
+            [Content(parts=[Part.from_text(text="Use a tool")], role="user")],
+            streaming_tool_names={"lookup"},
+        )
+        events = [
+            {"type": "response.created", "response": {"id": "resp_stream"}},
+            {"type": "response.reasoning_summary_text.delta", "delta": "Thinking"},
+            {"type": "response.output_text.delta", "delta": "Working"},
+            {
+                "type": "response.output_item.added",
+                "item": {
+                    "type": "function_call",
+                    "id": "fc_1",
+                    "call_id": "call_1",
+                    "name": "lookup",
+                    "arguments": "",
+                },
+            },
+            {"type": "response.function_call_arguments.delta", "item_id": "fc_1", "delta": '{"q":'},
+            {"type": "response.function_call_arguments.delta", "item_id": "fc_1", "delta": '"x"}'},
+            {
+                "type": "response.completed",
+                "response": {
+                    "id": "resp_stream",
+                    "status": "completed",
+                    "output": [
+                        {"type": "reasoning", "summary": [{"type": "summary_text", "text": "Thinking"}]},
+                        {
+                            "type": "message",
+                            "content": [{"type": "output_text", "text": "Working"}],
+                        },
+                        {
+                            "type": "function_call",
+                            "id": "fc_1",
+                            "call_id": "call_1",
+                            "name": "lookup",
+                            "arguments": '{"q":"x"}',
+                        },
+                    ],
+                    "usage": {"input_tokens": 5, "output_tokens": 4, "total_tokens": 9},
+                },
+            },
+        ]
+
+        async def stream_events():
+            for event in events:
+                yield event
+
+        with patch.object(model, "_create_async_client") as client_factory:
+            client = AsyncMock()
+            client.responses.create = AsyncMock(return_value=stream_events())
+            client_factory.return_value = client
+            responses = [item async for item in model.generate_async(request, stream=True)]
+
+        assert responses[0].content.parts[0].thought is True
+        assert responses[0].content.parts[0].text == "Thinking"
+        assert responses[1].content.parts[0].text == "Working"
+        tool_deltas = [item.content.parts[0].function_call.args["tool_streaming_args"] for item in responses[2:-1]]
+        assert tool_deltas == ['{"q":', '"x"}']
+        final = responses[-1]
+        assert final.partial is False
+        assert final.response_id == "resp_stream"
+        assert final.content.parts[-1].function_call.args == {"q": "x"}
+        assert final.usage_metadata.total_token_count == 9
+
+    @pytest.mark.asyncio
+    async def test_streaming_error_event_is_not_reported_as_success(self):
+        model = _model(use_responses_api=True)
+        request = _request([Content(parts=[Part.from_text(text="Hello")], role="user")])
+
+        async def stream_events():
+            yield {
+                "type": "error",
+                "response_id": "resp_error",
+                "code": "server_error",
+                "message": "upstream unavailable",
+            }
+
+        with patch.object(model, "_create_async_client") as client_factory:
+            client = AsyncMock()
+            client.responses.create = AsyncMock(return_value=stream_events())
+            client_factory.return_value = client
+            responses = [item async for item in model.generate_async(request, stream=True)]
+
+        assert responses[-1].response_id == "resp_error"
+        assert responses[-1].error_code == "server_error"
+        assert responses[-1].error_message == "upstream unavailable"
+
+    @pytest.mark.asyncio
+    async def test_streaming_uses_arguments_done_payload_in_fallback(self):
+        model = _model(use_responses_api=True)
+        request = _request([Content(parts=[Part.from_text(text="Use a tool")], role="user")])
+
+        async def stream_events():
+            yield {
+                "type": "response.output_item.added",
+                "item": {
+                    "type": "function_call",
+                    "id": "fc_1",
+                    "call_id": "call_1",
+                    "name": "lookup",
+                    "arguments": "",
+                },
+            }
+            yield {
+                "type": "response.function_call_arguments.done",
+                "item_id": "fc_1",
+                "arguments": '{"query":"done"}',
+            }
+
+        with patch.object(model, "_create_async_client") as client_factory:
+            client = AsyncMock()
+            client.responses.create = AsyncMock(return_value=stream_events())
+            client_factory.return_value = client
+            responses = [item async for item in model.generate_async(request, stream=True)]
+
+        final = responses[-1]
+        assert final.partial is False
+        assert final.content.parts[0].function_call.id == "call_1"
+        assert final.content.parts[0].function_call.args == {"query": "done"}
+
+    def test_incomplete_response_maps_reason_to_error(self):
+        model = _model(use_responses_api=True)
+
+        response = model._create_responses_response(
+            {
+                "id": "resp_incomplete",
+                "status": "incomplete",
+                "incomplete_details": {
+                    "reason": "max_output_tokens",
+                },
+                "output": [],
+            }
+        )
+
+        assert response.error_code == "incomplete"
+        assert response.error_message == "max_output_tokens"
diff --git a/tests/tools/test_long_running_tool.py b/tests/tools/test_long_running_tool.py
index a32af212..626de441 100644
--- a/tests/tools/test_long_running_tool.py
+++ b/tests/tools/test_long_running_tool.py
@@ -1,57 +1,19 @@
-# Tencent is pleased to support the open source community by making tRPC-Agent-Python available.
-#
-# Copyright (C) 2026 Tencent. All rights reserved.
-#
-# tRPC-Agent-Python is licensed under Apache-2.0.
+from trpc_agent_sdk.tools import is_tool_execution_error
 
-from __future__ import annotations
 
-import pytest
+def test_provider_argument_parse_error_is_not_a_long_running_success():
+    assert is_tool_execution_error(
+        {
+            "result": "An error occurred while parsing tool arguments. Please try again with valid JSON.",
+        }
+    )
 
-from trpc_agent_sdk.tools._long_running_tool import LongRunningFunctionTool
 
+def test_normal_tool_result_is_not_classified_as_an_invocation_error():
+    assert not is_tool_execution_error({"status": "pending", "questions": []})
+    assert not is_tool_execution_error({"status": "success", "value": True})
 
-def long_func(query: str) -> str:
-    """A long running function."""
-    return f"result-{query}"
 
-
-def long_func_no_doc(query: str) -> str:
-    return f"result-{query}"
-
-
-long_func_no_doc.__doc__ = None
-
-
-class TestLongRunningFunctionToolInit:
-
-    def test_init(self):
-        tool = LongRunningFunctionTool(long_func)
-        assert tool.is_long_running is True
-        assert tool.name == "long_func"
-
-    def test_init_with_invalid_filters_raises(self):
-        with pytest.raises(ValueError, match="not found"):
-            LongRunningFunctionTool(long_func, filters_name=["nonexistent"])
-
-
-class TestLongRunningFunctionToolGetDeclaration:
-
-    def test_declaration_appends_note(self):
-        tool = LongRunningFunctionTool(long_func)
-        decl = tool._get_declaration()
-        assert decl is not None
-        assert "long-running operation" in decl.description
-        assert "A long running function." in decl.description
-
-    def test_declaration_with_no_existing_description(self):
-        tool = LongRunningFunctionTool(long_func_no_doc)
-        decl = tool._get_declaration()
-        assert decl is not None
-        # When description is empty string '', the instruction is set with lstrip
-        assert "long-running operation" in (decl.description or "")
-
-    def test_declaration_contains_do_not_call_again(self):
-        tool = LongRunningFunctionTool(long_func)
-        decl = tool._get_declaration()
-        assert "Do not call this tool again" in decl.description
+def test_explicit_tool_errors_are_classified():
+    assert is_tool_execution_error({"status": "error", "error_message": "bad input"})
+    assert is_tool_execution_error({"error": "validation failed"})
diff --git a/tests/trpc_agent_dsl/graph/test_agent_node_hitl.py b/tests/trpc_agent_dsl/graph/test_agent_node_hitl.py
new file mode 100644
index 00000000..829e7e7b
--- /dev/null
+++ b/tests/trpc_agent_dsl/graph/test_agent_node_hitl.py
@@ -0,0 +1,291 @@
+# Tencent is pleased to support the open source community by making tRPC-Agent-Python available.
+#
+# Copyright (C) 2026 Tencent. All rights reserved.
+#
+# tRPC-Agent-Python is licensed under Apache-2.0.
+"""End-to-end HITL propagation tests for GraphAgent agent nodes."""
+
+from typing import AsyncGenerator
+from typing import List
+
+import pytest
+from trpc_agent_sdk.agents import BaseAgent
+from trpc_agent_sdk.dsl.graph import END
+from trpc_agent_sdk.dsl.graph import START
+from trpc_agent_sdk.dsl.graph import GraphAgent
+from trpc_agent_sdk.dsl.graph import State
+from trpc_agent_sdk.dsl.graph import StateGraph
+from trpc_agent_sdk.events import Event
+from trpc_agent_sdk.events import LongRunningEvent
+from trpc_agent_sdk.models import LLMModel
+from trpc_agent_sdk.models import LlmRequest
+from trpc_agent_sdk.models import LlmResponse
+from trpc_agent_sdk.runners import Runner
+from trpc_agent_sdk.sessions import InMemorySessionService
+from trpc_agent_sdk.sessions import SqlSessionService
+from trpc_agent_sdk.teams import TeamAgent
+from trpc_agent_sdk.teams.core import TEAM_STATE_KEY
+from trpc_agent_sdk.types import Content
+from trpc_agent_sdk.types import FunctionCall
+from trpc_agent_sdk.types import FunctionResponse
+from trpc_agent_sdk.types import Part
+
+
+class HitlState(State, total=False):
+    after_child: bool
+
+
+class StubModel(LLMModel):
+    @classmethod
+    def supported_models(cls) -> List[str]:
+        return [r"stub-model"]
+
+    async def _generate_async_impl(
+        self,
+        request: LlmRequest,
+        stream: bool = False,
+        ctx=None,
+    ) -> AsyncGenerator[LlmResponse, None]:
+        del request, stream, ctx
+        yield LlmResponse(content=None)
+
+    def validate_request(self, request: LlmRequest) -> None:
+        del request
+
+
+class TwoRoundClarifyingAgent(BaseAgent):
+    """Child agent that requires two FunctionResponses before completing."""
+
+    received_call_ids: list[str] = []
+
+    async def _run_async_impl(self, ctx) -> AsyncGenerator[Event, None]:
+        response = self._function_response(ctx.user_content)
+        if response is not None:
+            self.received_call_ids.append(str(response.id))
+            if response.response.get("answer") == "done":
+                yield Event(
+                    invocation_id=ctx.invocation_id,
+                    author=self.name,
+                    branch=ctx.branch,
+                    content=Content(role="model", parts=[Part.from_text(text="clarification-complete")]),
+                )
+                return
+            call_id = "child-question-2"
+        else:
+            call_id = "child-question-1"
+
+        call = FunctionCall(
+            id=call_id,
+            name="ask_clarification",
+            args={"round": 1 if call_id.endswith("1") else 2},
+        )
+        yield LongRunningEvent(
+            invocation_id=ctx.invocation_id,
+            author=self.name,
+            branch=ctx.branch,
+            function_call=call,
+            function_response=FunctionResponse(
+                id=call.id,
+                name=call.name,
+                response={"status": "pending"},
+            ),
+        )
+
+    @staticmethod
+    def _function_response(content: Content | None) -> FunctionResponse | None:
+        if content is None:
+            return None
+        for part in content.parts or []:
+            if part.function_response is not None:
+                return part.function_response
+        return None
+
+
+class TeamHitlLeader(BaseAgent):
+    """Leader double that verifies TeamAgent receives the original response ID."""
+
+    received_call_ids: list[str] = []
+
+    async def _run_async_impl(self, ctx) -> AsyncGenerator[Event, None]:
+        response = TwoRoundClarifyingAgent._function_response(ctx.user_content)
+        if response is not None:
+            self.received_call_ids.append(str(response.id))
+            yield Event(
+                invocation_id=ctx.invocation_id,
+                author=self.name,
+                branch=ctx.branch,
+                content=Content(role="model", parts=[Part.from_text(text="team-clarification-complete")]),
+            )
+            return
+
+        call = FunctionCall(
+            id="team-question-1",
+            name="ask_clarification",
+            args={"question": "approve team plan?"},
+        )
+        yield LongRunningEvent(
+            invocation_id=ctx.invocation_id,
+            author=self.name,
+            branch=ctx.branch,
+            function_call=call,
+            function_response=FunctionResponse(
+                id=call.id,
+                name=call.name,
+                response={"status": "pending"},
+            ),
+        )
+
+
+def _user_text(text: str) -> Content:
+    return Content(role="user", parts=[Part.from_text(text=text)])
+
+
+def _tool_response(event: LongRunningEvent, answer: str) -> Content:
+    return Content(
+        role="user",
+        parts=[
+            Part(
+                function_response=FunctionResponse(
+                    id=event.function_call.id,
+                    name=event.function_call.name,
+                    response={"answer": answer},
+                )
+            )
+        ],
+    )
+
+
+async def _run(runner: Runner, message: Content) -> list[Event]:
+    return [
+        event
+        async for event in runner.run_async(
+            user_id="user-1",
+            session_id="session-1",
+            new_message=message,
+        )
+    ]
+
+
+@pytest.mark.asyncio
+async def test_agent_node_long_running_interrupts_parent_and_resumes_multiple_rounds():
+    child = TwoRoundClarifyingAgent(name="clarifier")
+
+    async def after_child(state: HitlState) -> dict[str, bool]:
+        del state
+        return {"after_child": True}
+
+    graph = StateGraph(HitlState)
+    graph.add_agent_node("clarify", child, isolated_messages=True)
+    graph.add_node("after_child", after_child)
+    graph.add_edge(START, "clarify")
+    graph.add_edge("clarify", "after_child")
+    graph.add_edge("after_child", END)
+
+    service = InMemorySessionService()
+    runner = Runner(
+        app_name="agent-node-hitl-test",
+        agent=GraphAgent(name="workflow", graph=graph.compile()),
+        session_service=service,
+        close_session_service_on_close=False,
+    )
+
+    first_events = await _run(runner, _user_text("start"))
+    first_pending = next(event for event in first_events if isinstance(event, LongRunningEvent))
+    assert first_pending.function_call.name == "ask_clarification"
+    assert first_pending.function_call.args == {"round": 1, "status": "pending"}
+    first_session = await service.get_session(
+        app_name="agent-node-hitl-test",
+        user_id="user-1",
+        session_id="session-1",
+    )
+    assert first_session is not None
+    assert first_session.state.get("after_child") is not True
+
+    second_events = await _run(runner, _tool_response(first_pending, "again"))
+    second_pending = next(event for event in second_events if isinstance(event, LongRunningEvent))
+    assert second_pending.function_call.name == "ask_clarification"
+    second_session = await service.get_session(
+        app_name="agent-node-hitl-test",
+        user_id="user-1",
+        session_id="session-1",
+    )
+    assert second_session is not None
+    assert second_session.state.get("after_child") is not True
+
+    await _run(runner, _tool_response(second_pending, "done"))
+    completed_session = await service.get_session(
+        app_name="agent-node-hitl-test",
+        user_id="user-1",
+        session_id="session-1",
+    )
+    assert completed_session is not None
+    assert completed_session.state.get("after_child") is True
+    assert child.received_call_ids == ["child-question-1", "child-question-2"]
+
+    await runner.close()
+    await service.close()
+
+
+@pytest.mark.asyncio
+async def test_team_agent_leader_hitl_survives_service_restart(tmp_path):
+    async def after_team(state: HitlState) -> dict[str, bool]:
+        del state
+        return {"after_child": True}
+
+    def build_runner(service: SqlSessionService) -> tuple[Runner, TeamHitlLeader]:
+        member = TwoRoundClarifyingAgent(name="unused_member")
+        team = TeamAgent(
+            name="development_team",
+            model=StubModel(model_name="stub-model"),
+            members=[member],
+        )
+        leader = TeamHitlLeader(name="development_team")
+        team.__pydantic_private__["_leader_agent"] = leader
+        graph = StateGraph(HitlState)
+        graph.add_agent_node("development", team, isolated_messages=True)
+        graph.add_node("after_team", after_team)
+        graph.add_edge(START, "development")
+        graph.add_edge("development", "after_team")
+        graph.add_edge("after_team", END)
+        return (
+            Runner(
+                app_name="agent-node-hitl-test",
+                agent=GraphAgent(name="team_workflow", graph=graph.compile()),
+                session_service=service,
+                close_session_service_on_close=False,
+            ),
+            leader,
+        )
+
+    db_url = f"sqlite:///{tmp_path / 'agent-node-hitl.sqlite'}"
+    service = SqlSessionService(db_url, is_async=False)
+    runner, _ = build_runner(service)
+
+    first_events = await _run(runner, _user_text("start team"))
+    pending = next(event for event in first_events if isinstance(event, LongRunningEvent))
+    first_session = await service.get_session(
+        app_name="agent-node-hitl-test",
+        user_id="user-1",
+        session_id="session-1",
+    )
+    assert first_session is not None
+    assert first_session.state.get("after_child") is not True
+    assert TEAM_STATE_KEY in first_session.state
+
+    await runner.close()
+    await service.close()
+
+    service = SqlSessionService(db_url, is_async=False)
+    runner, resumed_leader = build_runner(service)
+    await _run(runner, _tool_response(pending, "done"))
+    completed_session = await service.get_session(
+        app_name="agent-node-hitl-test",
+        user_id="user-1",
+        session_id="session-1",
+    )
+    assert completed_session is not None
+    assert completed_session.state.get("after_child") is True
+    assert resumed_leader.received_call_ids == ["team-question-1"]
+
+    await runner.close()
+    await service.close()
diff --git a/trpc_agent_sdk/agents/_llm_agent.py b/trpc_agent_sdk/agents/_llm_agent.py
index 3546c1d6..66741d86 100644
--- a/trpc_agent_sdk/agents/_llm_agent.py
+++ b/trpc_agent_sdk/agents/_llm_agent.py
@@ -12,52 +12,43 @@
 
 from __future__ import annotations
 
-from typing import Any
-from typing import AsyncGenerator
-from typing import Awaitable
-from typing import Callable
-from typing import List
-from typing import Literal
-from typing import Optional
-from typing import TypeAlias
-from typing import Union
-from typing_extensions import override
+from typing import Any, AsyncGenerator, Awaitable, Callable, List, Literal, Optional, TypeAlias, Union
 
-from pydantic import BaseModel
-from pydantic import Field
-from pydantic import field_validator
+from pydantic import BaseModel, Field, field_validator
+from typing_extensions import override
 
 from trpc_agent_sdk.code_executors import BaseCodeExecutor
 from trpc_agent_sdk.context import InvocationContext
 from trpc_agent_sdk.events import Event
 from trpc_agent_sdk.log import logger
-from trpc_agent_sdk.models import LLMModel
-from trpc_agent_sdk.models import LlmRequest
-from trpc_agent_sdk.models import ModelRegistry
+from trpc_agent_sdk.models import LLMModel, LlmRequest, ModelRegistry
 from trpc_agent_sdk.planners import BasePlanner
 from trpc_agent_sdk.skills import BaseSkillRepository
-from trpc_agent_sdk.tools import BaseTool
-from trpc_agent_sdk.tools import BaseToolSet
-from trpc_agent_sdk.tools import FunctionTool
-from trpc_agent_sdk.tools import LongRunningFunctionTool
-from trpc_agent_sdk.tools import transfer_to_agent
-from trpc_agent_sdk.types import Content
-from trpc_agent_sdk.types import GenerateContentConfig
+from trpc_agent_sdk.tools import (
+    BaseTool,
+    BaseToolSet,
+    FunctionTool,
+    LongRunningFunctionTool,
+    is_tool_execution_error,
+    transfer_to_agent,
+)
+from trpc_agent_sdk.types import Content, GenerateContentConfig
 
 from ..exceptions import RunCancelledException
 from ._base_agent import BaseAgent
-from ._callback import ModelCallback
-from ._callback import ToolCallback
+from ._callback import ModelCallback, ToolCallback
 from ._constants import TRPC_AGENT_RUNNING_KEY
-from .core import BranchFilterMode
-from .core import CodeExecutionRequestProcessor
-from .core import CodeExecutionResponseProcessor
-from .core import LlmProcessor
-from .core import TimelineFilterMode
-from .core import ToolsProcessor
-from .core import create_final_model_response_event
-from .core import default_request_processor
-from .core import get_structured_model_response
+from .core import (
+    BranchFilterMode,
+    CodeExecutionRequestProcessor,
+    CodeExecutionResponseProcessor,
+    LlmProcessor,
+    TimelineFilterMode,
+    ToolsProcessor,
+    create_final_model_response_event,
+    default_request_processor,
+    get_structured_model_response,
+)
 
 # Type aliases for instruction providers
 InstructionProvider: TypeAlias = Callable[[InvocationContext], Union[str, Awaitable[str]]]
@@ -118,7 +109,7 @@ class LlmAgent(BaseAgent):
     settings, etc.
     """
 
-    include_contents: Literal['default', 'none'] = 'default'
+    include_contents: Literal["default", "none"] = "default"
     """Controls content inclusion in model requests.
 
     Options:
@@ -368,8 +359,12 @@ def _should_enable_agent_transfer(self) -> bool:
             return True
 
         # Check if transfer to peers is possible
-        if (not self.disallow_transfer_to_peers and self.parent_agent and self._is_llm_agent(self.parent_agent)
-                and len(self.parent_agent.sub_agents) > 1):  # Has siblings
+        if (
+            not self.disallow_transfer_to_peers
+            and self.parent_agent
+            and self._is_llm_agent(self.parent_agent)
+            and len(self.parent_agent.sub_agents) > 1
+        ):  # Has siblings
             return True
 
         return False
@@ -471,7 +466,9 @@ def accumulate_content(event: Event) -> None:
                 await ctx.raise_if_cancelled()
 
                 # Step 1: Build request using the request processor (includes conversation history)
-                request = LlmRequest(model=model_instance.name, )
+                request = LlmRequest(
+                    model=model_instance.name,
+                )
 
                 error_event = await default_request_processor.build_request(
                     request,
@@ -579,8 +576,7 @@ def accumulate_content(event: Event) -> None:
                             # Check if this event contains responses from long-running tools
                             if tool_event.content and tool_event.content.parts:
                                 for part in tool_event.content.parts:
-                                    if (part.function_response and part.function_response.id in long_running_tool_ids):
-                                        # This is a response from a long-running tool
+                                    if part.function_response and part.function_response.id in long_running_tool_ids:
                                         # Find the corresponding function call
                                         corresponding_call = None
                                         for call in collected_tool_calls:
@@ -589,6 +585,16 @@ def accumulate_content(event: Event) -> None:
                                                 break
 
                                         if corresponding_call:
+                                            if is_tool_execution_error(part.function_response.response):
+                                                logger.warning(
+                                                    "Long-running tool %s returned an invocation error; "
+                                                    "continuing as a normal tool response",
+                                                    corresponding_call.name,
+                                                )
+                                                # The model can see the error and decide whether to
+                                                # correct and retry. Never suspend the graph here.
+                                                continue
+                                            # This is a response from a long-running tool
                                             # Import LongRunningEvent here to avoid circular imports
                                             from trpc_agent_sdk.events import LongRunningEvent
 
@@ -607,8 +613,10 @@ def accumulate_content(event: Event) -> None:
                                             # Then yield the long-running event
                                             yield long_running_event
 
-                                            logger.debug("Long-running tool %s completed, yielding LongRunningEvent",
-                                                         corresponding_call.name)
+                                            logger.debug(
+                                                "Long-running tool %s completed, yielding LongRunningEvent",
+                                                corresponding_call.name,
+                                            )
                                             return  # End agent execution after long-running event
 
                             # Yield regular tool events
@@ -701,33 +709,34 @@ def _save_output_to_state(self, ctx: InvocationContext, event: Event) -> None:
         """
         if self.output_key:
             # Save output to state using the delta tracking system
-            result = ''.join([part.text if not part.thought else '' for part in event.content.parts])
+            result = "".join([part.text if not part.thought else "" for part in event.content.parts])
             ctx.state[self.output_key] = result
             event.actions.state_delta[self.output_key] = result
             logger.debug("Saved agent output to state key '%s': %s...", self.output_key, result[:100])
 
-    @field_validator('generate_content_config', mode='after')
+    @field_validator("generate_content_config", mode="after")
     @classmethod
     def __validate_generate_content_config(
-            cls, generate_content_config: Optional[GenerateContentConfig]) -> GenerateContentConfig:
+        cls, generate_content_config: Optional[GenerateContentConfig]
+    ) -> GenerateContentConfig:
         if not generate_content_config:
             return GenerateContentConfig()
         if generate_content_config.thinking_config:
-            raise ValueError('Thinking config should be set via LlmAgent.planner.')
+            raise ValueError("Thinking config should be set via LlmAgent.planner.")
         if generate_content_config.tools:
-            raise ValueError('All tools must be set via LlmAgent.tools.')
+            raise ValueError("All tools must be set via LlmAgent.tools.")
         if generate_content_config.system_instruction:
-            raise ValueError('System instruction must be set via LlmAgent.instruction.')
+            raise ValueError("System instruction must be set via LlmAgent.instruction.")
         if generate_content_config.response_schema:
-            raise ValueError('Response schema must be set via LlmAgent.output_schema.')
+            raise ValueError("Response schema must be set via LlmAgent.output_schema.")
         return generate_content_config
 
-    @field_validator('code_executor', mode='after')
+    @field_validator("code_executor", mode="after")
     @classmethod
     def __validate_code_executor(cls, code_executor: Optional[BaseCodeExecutor]) -> Optional[BaseCodeExecutor]:
         """Validate code executor configuration."""
         if code_executor and not isinstance(code_executor, BaseCodeExecutor):
-            raise ValueError('Code executor must be an instance of BaseCodeExecutor.')
+            raise ValueError("Code executor must be an instance of BaseCodeExecutor.")
         return code_executor
 
 
diff --git a/trpc_agent_sdk/dsl/graph/_constants.py b/trpc_agent_sdk/dsl/graph/_constants.py
index c2d04072..04393835 100644
--- a/trpc_agent_sdk/dsl/graph/_constants.py
+++ b/trpc_agent_sdk/dsl/graph/_constants.py
@@ -148,6 +148,11 @@
 STATE_KEY_PENDING_INTERRUPT_AUTHOR = "_trpc_graph_pending_interrupt_author"
 STATE_KEY_PENDING_INTERRUPT_BRANCH = "_trpc_graph_pending_interrupt_branch"
 
+# Pending child-agent HITL context owned by AgentNodeAction. This bridges a
+# child LongRunningEvent into a parent GraphAgent interrupt and supports
+# replaying multiple clarification rounds in the same graph node.
+STATE_KEY_PENDING_AGENT_NODE_HITL = "_trpc_graph_pending_agent_node_hitl"
+
 # Prefix for synthetic function call IDs used by graph interrupt bridge
 STATE_KEY_LONG_RUNNING_PREFIX = "__trpc_graph_long_running__"
 
@@ -164,22 +169,24 @@
 # Unsafe State Keys (not serializable or should not be exposed)
 # =============================================================================
 
-UNSAFE_STATE_KEYS = frozenset({
-    STATE_KEY_SESSION,
-    STATE_KEY_EXEC_CONTEXT,
-    STATE_KEY_CURRENT_NODE_ID,
-    STATE_KEY_TOOL_CALLBACKS,
-    STATE_KEY_MODEL_CALLBACKS,
-    STATE_KEY_AGENT_CALLBACKS,
-    STATE_KEY_NODE_CALLBACKS,
-    STATE_KEY_CHECKPOINTS,
-    STATE_KEY_CHECKPOINT_WRITES,
-    STATE_KEY_CHECKPOINT_BLOBS,
-    STATE_KEY_PENDING_INTERRUPT,
-    STATE_KEY_PENDING_INTERRUPT_ID,
-    STATE_KEY_PENDING_INTERRUPT_AUTHOR,
-    STATE_KEY_PENDING_INTERRUPT_BRANCH,
-})
+UNSAFE_STATE_KEYS = frozenset(
+    {
+        STATE_KEY_SESSION,
+        STATE_KEY_EXEC_CONTEXT,
+        STATE_KEY_CURRENT_NODE_ID,
+        STATE_KEY_TOOL_CALLBACKS,
+        STATE_KEY_MODEL_CALLBACKS,
+        STATE_KEY_AGENT_CALLBACKS,
+        STATE_KEY_NODE_CALLBACKS,
+        STATE_KEY_CHECKPOINTS,
+        STATE_KEY_CHECKPOINT_WRITES,
+        STATE_KEY_CHECKPOINT_BLOBS,
+        STATE_KEY_PENDING_INTERRUPT,
+        STATE_KEY_PENDING_INTERRUPT_ID,
+        STATE_KEY_PENDING_INTERRUPT_AUTHOR,
+        STATE_KEY_PENDING_INTERRUPT_BRANCH,
+    }
+)
 
 
 def is_unsafe_state_key(key: str) -> bool:
diff --git a/trpc_agent_sdk/dsl/graph/_graph_agent.py b/trpc_agent_sdk/dsl/graph/_graph_agent.py
index 12e05c1d..3d484720 100644
--- a/trpc_agent_sdk/dsl/graph/_graph_agent.py
+++ b/trpc_agent_sdk/dsl/graph/_graph_agent.py
@@ -113,8 +113,10 @@ class GraphAgent(BaseAgent):
     @classmethod
     def _reject_sub_agents(cls, data: Any) -> Any:
         if isinstance(data, dict) and "sub_agents" in data:
-            raise ValueError("GraphAgent does not accept `sub_agents`; compose nested agents with "
-                             "`StateGraph.add_agent_node(...)`.")
+            raise ValueError(
+                "GraphAgent does not accept `sub_agents`; compose nested agents with "
+                "`StateGraph.add_agent_node(...)`."
+            )
         return data
 
     @override
@@ -178,9 +180,9 @@ async def _run_async_impl(self, ctx: InvocationContext) -> AsyncGenerator[Event,
             # "updates" is required for state to propagate between nodes
             # "custom" enables EventEmitter streaming
             async for mode, chunk in self.graph.astream(
-                    graph_input,
-                    runnable_config,
-                    stream_mode=["updates", "custom"],
+                graph_input,
+                runnable_config,
+                stream_mode=["updates", "custom"],
             ):
                 # Cancellation checkpoint at each iteration
                 await ctx.raise_if_cancelled()
@@ -193,7 +195,8 @@ async def _run_async_impl(self, ctx: InvocationContext) -> AsyncGenerator[Event,
                     interrupted = True
                     for interrupt in interrupts:
                         function_call_event, function_response_event, long_running_event = (
-                            self._create_interrupt_events(ctx, interrupt))
+                            self._create_interrupt_events(ctx, interrupt)
+                        )
                         yield function_call_event
                         await ctx.raise_if_cancelled()
                         yield function_response_event
@@ -216,7 +219,8 @@ async def _run_async_impl(self, ctx: InvocationContext) -> AsyncGenerator[Event,
                                     if isinstance(key, str):
                                         final_state[key] = value
                                 logger.debug(
-                                    f"[{self.name}] Node '{node_name}' updated keys: {list(node_output.keys())}")
+                                    f"[{self.name}] Node '{node_name}' updated keys: {list(node_output.keys())}"
+                                )
 
                         # Emit state update event
                         if updated_keys:
@@ -293,7 +297,7 @@ def _extract_resume_command(self, ctx: InvocationContext) -> Optional[Command]:
 
         if not function_response_id.startswith(STATE_KEY_LONG_RUNNING_PREFIX):
             return None
-        interrupt_id = function_response_id[len(STATE_KEY_LONG_RUNNING_PREFIX):]
+        interrupt_id = function_response_id.removeprefix(STATE_KEY_LONG_RUNNING_PREFIX)
         if not interrupt_id:
             return None
 
@@ -390,7 +394,11 @@ def _build_interrupt_function(self, interrupt: Interrupt) -> tuple[str, str, dic
         function_call_id = f"{STATE_KEY_LONG_RUNNING_PREFIX}{interrupt_id}"
 
         raw_args = interrupt.value
-        if isinstance(raw_args, dict):
+        if isinstance(raw_args, dict) and raw_args.get("_trpc_agent_node_hitl") is True:
+            function_name = str(raw_args.get("toolName") or function_name)
+            visible_args = raw_args.get("arguments")
+            function_args = visible_args if isinstance(visible_args, dict) else {}
+        elif isinstance(raw_args, dict):
             function_args = {str(key): value for key, value in raw_args.items()}
         else:
             function_args = {"value": raw_args}
@@ -447,15 +455,17 @@ def _build_initial_state(self, ctx: InvocationContext) -> GraphState:
         else:
             initial_state[STATE_KEY_MESSAGES] = messages
 
-        initial_state.update({
-            STATE_KEY_USER_INPUT: user_input,
-            STATE_KEY_METADATA: {
-                METADATA_KEY_INVOCATION_ID: ctx.invocation_id,
-                METADATA_KEY_BRANCH: ctx.branch or self.name,
-                METADATA_KEY_SESSION_ID: ctx.session.id,
-                METADATA_KEY_AGENT_NAME: self.name,
-            },
-        })
+        initial_state.update(
+            {
+                STATE_KEY_USER_INPUT: user_input,
+                STATE_KEY_METADATA: {
+                    METADATA_KEY_INVOCATION_ID: ctx.invocation_id,
+                    METADATA_KEY_BRANCH: ctx.branch or self.name,
+                    METADATA_KEY_SESSION_ID: ctx.session.id,
+                    METADATA_KEY_AGENT_NAME: self.name,
+                },
+            }
+        )
 
         # Built-ins should always exist for node logic consistency.
         initial_state.setdefault(STATE_KEY_LAST_RESPONSE, None)
diff --git a/trpc_agent_sdk/dsl/graph/_node_action/_agent.py b/trpc_agent_sdk/dsl/graph/_node_action/_agent.py
index 85d064dc..5bc7ed9b 100644
--- a/trpc_agent_sdk/dsl/graph/_node_action/_agent.py
+++ b/trpc_agent_sdk/dsl/graph/_node_action/_agent.py
@@ -10,12 +10,15 @@
 from typing import Callable
 from typing import Optional
 
+from langgraph.errors import GraphInterrupt
 from trpc_agent_sdk.agents import BaseAgent
 from trpc_agent_sdk.agents import LlmAgent
 from trpc_agent_sdk.context import InvocationContext
 from trpc_agent_sdk.events import Event
+from trpc_agent_sdk.events import LongRunningEvent
 from trpc_agent_sdk.types import Content
 from trpc_agent_sdk.types import EventActions
+from trpc_agent_sdk.types import FunctionResponse
 from trpc_agent_sdk.types import Part
 
 from .._callbacks import NodeCallbackContext
@@ -23,9 +26,11 @@
 from .._constants import STATE_KEY_LAST_RESPONSE
 from .._constants import STATE_KEY_MESSAGES
 from .._constants import STATE_KEY_NODE_RESPONSES
+from .._constants import STATE_KEY_PENDING_AGENT_NODE_HITL
 from .._constants import STATE_KEY_USER_INPUT
 from .._event_writer import AsyncEventWriter
 from .._event_writer import EventWriter
+from .._interrupt import interrupt
 from .._node_config import NodeConfig
 from .._state import State
 from .._state_mapper import SubgraphResult
@@ -117,22 +122,45 @@ async def execute(self, state: State) -> dict[str, Any]:
         if parent_ctx is None:
             raise RuntimeError(
                 f"Agent node '{self.name}' requires InvocationContext but none was set. "
-                "Pass context via config['configurable']['invocation_context'] when executing the graph.")
+                "Pass context via config['configurable']['invocation_context'] when executing the graph."
+            )
 
         child_scope = self.event_scope or self.agent.name
         child_branch = f"{parent_ctx.branch}.{child_scope}" if parent_ctx.branch else child_scope
         child_user_input = child_state.get(STATE_KEY_USER_INPUT, "")
 
+        pending_hitl = self._get_pending_hitl(parent_ctx)
+        resume_content: Optional[Content] = None
+        if pending_hitl is not None:
+            completed_rounds = pending_hitl.get("completed", [])
+            for completed in completed_rounds if isinstance(completed_rounds, list) else []:
+                if isinstance(completed, dict):
+                    interrupt(self._interrupt_payload(completed))
+
+            current_round = pending_hitl.get("current")
+            if not isinstance(current_round, dict):
+                raise RuntimeError(f"Agent node '{self.node_id}' has invalid pending HITL state.")
+            human_response = interrupt(self._interrupt_payload(current_round))
+            resume_content = self._resume_content(current_round, human_response)
+            saved_child_state = current_round.get("child_state")
+            if isinstance(saved_child_state, dict):
+                child_state = dict(saved_child_state)
+
         child_session = parent_ctx.session.model_copy(deep=True)
         child_session.state = dict(child_state)
-        child_events = self._build_child_events(parent_ctx, child_user_input, child_branch)
+        child_events = self._build_child_events(
+            parent_ctx,
+            child_user_input,
+            child_branch,
+            resume_content=resume_content,
+        )
         if hasattr(child_session, "events"):
             child_session.events = child_events
         if self.isolated_messages:
             child_session.state[STATE_KEY_MESSAGES] = []
 
-        child_user_content = None
-        if isinstance(child_user_input, str) and child_user_input:
+        child_user_content = resume_content
+        if child_user_content is None and isinstance(child_user_input, str) and child_user_input:
             child_user_content = Content(
                 role="user",
                 parts=[Part.from_text(text=child_user_input)],
@@ -164,11 +192,35 @@ async def execute(self, state: State) -> dict[str, Any]:
                 transfer_requested = False
                 child_ctx.agent = current_agent
 
-                async for event in current_agent.run_async(child_ctx):
+                agent_stream = current_agent.run_async(child_ctx)
+                async for event in agent_stream:
                     await self._run_agent_event_callbacks(state, event)
 
+                    if isinstance(event, LongRunningEvent):
+                        current_round = self._pending_round(event, child_ctx, final_state)
+                        completed_rounds: list[dict[str, Any]] = []
+                        if pending_hitl is not None:
+                            previous = pending_hitl.get("completed", [])
+                            if isinstance(previous, list):
+                                completed_rounds.extend(item for item in previous if isinstance(item, dict))
+                            previous_current = pending_hitl.get("current")
+                            if isinstance(previous_current, dict):
+                                completed_rounds.append(
+                                    {key: value for key, value in previous_current.items() if key != "child_state"}
+                                )
+                        parent_ctx.state[STATE_KEY_PENDING_AGENT_NODE_HITL] = {
+                            "node_id": self.node_id,
+                            "completed": completed_rounds,
+                            "current": current_round,
+                        }
+                        await agent_stream.aclose()
+                        interrupt(self._interrupt_payload(current_round))
+                        raise RuntimeError(f"Agent node '{self.node_id}' did not suspend after LongRunningEvent.")
+
                     if (not event.partial) and hasattr(child_session, "events"):
-                        child_session.events.append(event.model_copy(deep=True))
+                        existing_ids = {item.id for item in child_session.events if getattr(item, "id", None)}
+                        if not event.id or event.id not in existing_ids:
+                            child_session.events.append(event.model_copy(deep=True))
 
                     if event.actions and event.actions.state_delta:
                         delta = dict(event.actions.state_delta)
@@ -244,7 +296,14 @@ async def execute(self, state: State) -> dict[str, Any]:
                 if isinstance(candidate, str) and candidate:
                     last_response = candidate
 
+            if pending_hitl is not None:
+                parent_ctx.state[STATE_KEY_PENDING_AGENT_NODE_HITL] = None
+
+        except GraphInterrupt:
+            raise
         except Exception as e:
+            if pending_hitl is not None:
+                parent_ctx.state[STATE_KEY_PENDING_AGENT_NODE_HITL] = None
             raise RuntimeError(f"Agent node '{self.name}' execution failed: {e}") from e
 
         if last_response:
@@ -265,9 +324,7 @@ async def execute(self, state: State) -> dict[str, Any]:
 
         default_result = {
             STATE_KEY_LAST_RESPONSE: last_response,
-            STATE_KEY_NODE_RESPONSES: {
-                self.node_id: node_response
-            },
+            STATE_KEY_NODE_RESPONSES: {self.node_id: node_response},
             STATE_KEY_USER_INPUT: "",
         }
 
@@ -276,8 +333,9 @@ async def execute(self, state: State) -> dict[str, Any]:
             if mapped is None:
                 return {}
             if not isinstance(mapped, dict):
-                raise TypeError(f"Output mapper for agent node '{self.node_id}' must return dict, "
-                                f"got {type(mapped).__name__}.")
+                raise TypeError(
+                    f"Output mapper for agent node '{self.node_id}' must return dict, " f"got {type(mapped).__name__}."
+                )
             if STATE_KEY_USER_INPUT not in mapped:
                 mapped = dict(mapped)
                 mapped[STATE_KEY_USER_INPUT] = ""
@@ -290,10 +348,19 @@ def _build_child_events(
         parent_ctx: InvocationContext,
         child_user_input: Any,
         child_branch: str,
+        *,
+        resume_content: Optional[Content] = None,
     ) -> list[Event]:
         parent_events = getattr(parent_ctx.session, "events", [])
-        if self.isolated_messages:
-            child_events: list[Event] = []
+        pending_hitl = self._get_pending_hitl(parent_ctx)
+        if self.isolated_messages and pending_hitl is not None:
+            child_events = [
+                event.model_copy(deep=True)
+                for event in parent_events
+                if event.branch == child_branch or str(event.branch or "").startswith(f"{child_branch}.")
+            ]
+        elif self.isolated_messages:
+            child_events = []
         else:
             child_events = [event.model_copy(deep=True) for event in parent_events]
 
@@ -307,9 +374,77 @@ def _build_child_events(
                         role="user",
                         parts=[Part.from_text(text=child_user_input)],
                     ),
-                ))
+                )
+            )
+        if resume_content is not None:
+            child_events.append(
+                Event(
+                    invocation_id=parent_ctx.invocation_id,
+                    author="user",
+                    branch=child_branch,
+                    content=resume_content.model_copy(deep=True),
+                )
+            )
         return child_events
 
+    def _get_pending_hitl(self, parent_ctx: InvocationContext) -> Optional[dict[str, Any]]:
+        value = parent_ctx.session.state.get(STATE_KEY_PENDING_AGENT_NODE_HITL)
+        if isinstance(value, dict) and value.get("node_id") == self.node_id:
+            return value
+        return None
+
+    def _pending_round(
+        self,
+        event: LongRunningEvent,
+        child_ctx: InvocationContext,
+        child_state: dict[str, Any],
+    ) -> dict[str, Any]:
+        return {
+            "agent_name": event.author or self.agent.name,
+            "branch": event.branch or child_ctx.branch,
+            "function_call": event.function_call.model_dump(mode="json"),
+            "function_response": event.function_response.model_dump(mode="json"),
+            "child_state": dict(child_state),
+        }
+
+    def _interrupt_payload(self, pending_round: dict[str, Any]) -> dict[str, Any]:
+        function_call = pending_round.get("function_call")
+        function_call = function_call if isinstance(function_call, dict) else {}
+        arguments = function_call.get("args")
+        arguments = arguments if isinstance(arguments, dict) else {}
+        function_response = pending_round.get("function_response")
+        function_response = function_response if isinstance(function_response, dict) else {}
+        response = function_response.get("response")
+        response = response if isinstance(response, dict) else {}
+        return {
+            "_trpc_agent_node_hitl": True,
+            "nodeId": self.node_id,
+            "agentName": str(pending_round.get("agent_name") or self.agent.name),
+            "toolName": str(function_call.get("name") or "graph_interrupt"),
+            # Long-running tools return the UI interaction contract from the
+            # tool execution. Preserve the model-supplied call arguments and
+            # let the tool result add derived fields such as stable IDs.
+            "arguments": {**arguments, **response},
+        }
+
+    @staticmethod
+    def _resume_content(pending_round: dict[str, Any], human_response: Any) -> Content:
+        function_call = pending_round.get("function_call")
+        function_call = function_call if isinstance(function_call, dict) else {}
+        response = human_response if isinstance(human_response, dict) else {"value": human_response}
+        return Content(
+            role="user",
+            parts=[
+                Part(
+                    function_response=FunctionResponse(
+                        id=str(function_call.get("id") or ""),
+                        name=str(function_call.get("name") or ""),
+                        response=response,
+                    )
+                )
+            ],
+        )
+
     async def _run_agent_event_callbacks(self, state: State, event: Event) -> None:
         if not self.callbacks or not self.callbacks.agent_event:
             return
diff --git a/trpc_agent_sdk/models/_openai_model.py b/trpc_agent_sdk/models/_openai_model.py
index bd30ae33..bb7c698a 100644
--- a/trpc_agent_sdk/models/_openai_model.py
+++ b/trpc_agent_sdk/models/_openai_model.py
@@ -110,6 +110,9 @@ class ApiParamsKey(str, Enum):
     PROMPT_CACHE_RETENTION = "prompt_cache_retention"
 
 
+_RESPONSES_INPUT_ITEMS = "responses_input_items"
+
+
 @register_model(model_name="OpenAIModel", supported_models=[r"gpt-.*", r"o1-.*", r"deepseek-.*", r"hy3-.*"])
 class OpenAIModel(LLMModel):
     """OpenAI model implementation using the abstract model interface.
@@ -131,6 +134,11 @@ class OpenAIModel(LLMModel):
                                 will be used as the base, with per-request configs
                                 overriding specific fields. Useful for maintaining
                                 consistent model behavior across multiple calls.
+        use_responses_api: Use ``client.responses.create`` instead of Chat Completions.
+                           Defaults to False for backward compatibility.
+        responses_api_params: Optional Responses-only parameters such as ``store``,
+                              ``reasoning``, ``include``, or ``truncation``. The model,
+                              input, and stream parameters remain managed by this class.
         **kwargs: Additional arguments passed to parent LLMModel class
                  (e.g., api_key, base_url, etc.)
 
@@ -168,6 +176,8 @@ def __init__(
         tool_prompt: str = "xml",
         generate_content_config: Optional[GenerateContentConfig] = None,
         http_client_provider_factory: HttpClientProviderFactory = temporary_http_client_provider_factory,
+        use_responses_api: bool = False,
+        responses_api_params: Optional[Dict[str, Any]] = None,
         **kwargs,
     ):
         super().__init__(model_name, filters_name, **kwargs)
@@ -176,6 +186,12 @@ def __init__(
         # Extract OpenAI-specific config
         self.organization: str = kwargs.get(const.ORGANIZATION, "")
         self.client_args = kwargs.get(const.CLIENT_ARGS, {})
+        self.use_responses_api = use_responses_api
+        self.responses_api_params = dict(responses_api_params or {})
+        reserved_response_params = {"model", "input", "stream"}.intersection(self.responses_api_params)
+        if reserved_response_params:
+            names = ", ".join(sorted(reserved_response_params))
+            raise ValueError(f"responses_api_params cannot override managed parameters: {names}")
         # Allow callers to inject a tuned httpx client
         http_client_provider_factory = http_client_provider_factory or temporary_http_client_provider_factory
         self._http_client_provider: BaseHttpClientProvider = http_client_provider_factory()
@@ -234,7 +250,7 @@ def _create_async_client(self) -> openai.AsyncOpenAI:
         logging.getLogger("httpx").setLevel(logging.WARNING)
 
         client_args = self.client_args.copy()
-        client_args['http_client'] = self._http_client_provider.create_http_client()
+        client_args["http_client"] = self._http_client_provider.create_http_client()
 
         return openai.AsyncOpenAI(
             api_key=self._api_key,
@@ -320,8 +336,13 @@ def _format_messages(self, request: LlmRequest) -> List[Dict[str, Any]]:
 
             parts: list[Part] = content.parts  # type: ignore
             conditions_iter = [
-                len(parts) == 1, parts[0].text, parts[0].function_call, parts[0].function_response,
-                parts[0].code_execution_result, parts[0].executable_code, parts[0].inline_data
+                len(parts) == 1,
+                parts[0].text,
+                parts[0].function_call,
+                parts[0].function_response,
+                parts[0].code_execution_result,
+                parts[0].executable_code,
+                parts[0].inline_data,
             ]
             # Handle different content structures
             if all(conditions_iter):
@@ -337,11 +358,21 @@ def _format_messages(self, request: LlmRequest) -> List[Dict[str, Any]]:
                 text_parts = []
                 image_parts = []
                 tool_calls = []
+                responses_input_items = []
 
                 for part in parts:  # type: ignore
+                    if part.thought:
+                        if self.use_responses_api and part.thought_signature:
+                            try:
+                                raw = part.thought_signature
+                                raw = raw.decode("utf-8") if isinstance(raw, bytes) else raw
+                                reasoning_item = json.loads(raw)
+                                if isinstance(reasoning_item, dict) and reasoning_item.get("type") == "reasoning":
+                                    responses_input_items.append(reasoning_item)
+                            except (UnicodeDecodeError, json.JSONDecodeError, TypeError):
+                                logger.warning("Ignoring invalid Responses reasoning item metadata")
+                        continue
                     if part.text:
-                        if part.thought:
-                            continue
                         text_parts.append(part.text)
                     elif part.inline_data and part.inline_data.mime_type:
                         # Handle image data - convert to OpenAI vision format
@@ -355,10 +386,12 @@ def _format_messages(self, request: LlmRequest) -> List[Dict[str, Any]]:
                                 "id": getattr(part.function_call, "id", None) or f"call_{uuid.uuid4().hex[:24]}",
                                 "type": "function",
                                 "function": {
-                                    "name":
-                                    part.function_call.name,
-                                    "arguments": (part.function_call.args if isinstance(part.function_call.args, str)
-                                                  else json.dumps(part.function_call.args, ensure_ascii=False)),
+                                    "name": part.function_call.name,
+                                    "arguments": (
+                                        part.function_call.args
+                                        if isinstance(part.function_call.args, str)
+                                        else json.dumps(part.function_call.args, ensure_ascii=False)
+                                    ),
                                 },
                             }
                             if self._adapter.should_include_thought_signature():
@@ -405,8 +438,11 @@ def _format_messages(self, request: LlmRequest) -> List[Dict[str, Any]]:
                 else:
                     for func_response in function_responses:
                         # Standard tool message format for OpenAI API
-                        raw_text = (json.dumps(func_response.response, ensure_ascii=False) if isinstance(
-                            func_response.response, dict) else str(func_response.response))
+                        raw_text = (
+                            json.dumps(func_response.response, ensure_ascii=False)
+                            if isinstance(func_response.response, dict)
+                            else str(func_response.response)
+                        )
                         clipped_text = self._clip_tool_response_text(
                             raw_text,
                             getattr(func_response, "name", "tool"),
@@ -448,6 +484,9 @@ def _format_messages(self, request: LlmRequest) -> List[Dict[str, Any]]:
                     if self._adapter.should_backfill_reasoning_content(role, message):
                         message[const.REASONING_CONTENT] = ""
 
+                    if responses_input_items:
+                        message[_RESPONSES_INPUT_ITEMS] = responses_input_items
+
                     formatted_messages.append(message)
 
         # Validate and fix message sequence for OpenAI compatibility
@@ -487,10 +526,9 @@ def _validate_and_fix_openai_messages(self, messages: List[Dict[str, Any]]) -> L
                             dummy_response = {
                                 const.ROLE: const.TOOL,
                                 const.TOOL_CALL_ID: pending_call["id"],
-                                const.CONTENT: json.dumps({
-                                    "status": "completed",
-                                    "note": "Tool call completed by system"
-                                }),
+                                const.CONTENT: json.dumps(
+                                    {"status": "completed", "note": "Tool call completed by system"}
+                                ),
                             }
                             fixed_messages.append(dummy_response)
 
@@ -503,16 +541,17 @@ def _validate_and_fix_openai_messages(self, messages: List[Dict[str, Any]]) -> L
                     # Assistant message without tool calls
                     if pending_tool_calls:
                         # Need to add dummy responses for pending tool calls
-                        logger.warning("Adding dummy tool responses for %s pending tool calls before assistant message",
-                                       len(pending_tool_calls))
+                        logger.warning(
+                            "Adding dummy tool responses for %s pending tool calls before assistant message",
+                            len(pending_tool_calls),
+                        )
                         for pending_call in pending_tool_calls:
                             dummy_response = {
                                 const.ROLE: const.TOOL,
                                 const.TOOL_CALL_ID: pending_call["id"],
-                                const.CONTENT: json.dumps({
-                                    "status": "completed",
-                                    "note": "Tool call completed by system"
-                                }),
+                                const.CONTENT: json.dumps(
+                                    {"status": "completed", "note": "Tool call completed by system"}
+                                ),
                             }
                             fixed_messages.append(dummy_response)
                         pending_tool_calls = []
@@ -533,16 +572,16 @@ def _validate_and_fix_openai_messages(self, messages: List[Dict[str, Any]]) -> L
                 # User or system message
                 if pending_tool_calls:
                     # Add dummy responses for any pending tool calls before user/system message
-                    logger.warning("Adding dummy tool responses for %s pending tool calls before %s message",
-                                   len(pending_tool_calls), role)
+                    logger.warning(
+                        "Adding dummy tool responses for %s pending tool calls before %s message",
+                        len(pending_tool_calls),
+                        role,
+                    )
                     for pending_call in pending_tool_calls:
                         dummy_response = {
                             const.ROLE: const.TOOL,
                             const.TOOL_CALL_ID: pending_call["id"],
-                            const.CONTENT: json.dumps({
-                                "status": "completed",
-                                "note": "Tool call completed by system"
-                            }),
+                            const.CONTENT: json.dumps({"status": "completed", "note": "Tool call completed by system"}),
                         }
                         fixed_messages.append(dummy_response)
                     pending_tool_calls = []
@@ -556,10 +595,7 @@ def _validate_and_fix_openai_messages(self, messages: List[Dict[str, Any]]) -> L
                 dummy_response = {
                     const.ROLE: const.TOOL,
                     const.TOOL_CALL_ID: pending_call["id"],
-                    const.CONTENT: json.dumps({
-                        "status": "completed",
-                        "note": "Tool call completed by system"
-                    }),
+                    const.CONTENT: json.dumps({"status": "completed", "note": "Tool call completed by system"}),
                 }
                 fixed_messages.append(dummy_response)
 
@@ -609,8 +645,10 @@ def _is_thinking_event(self, response: dict) -> bool:
         Returns:
             `bool`: True if this is a thinking event
         """
-        return (response.get("object", "") == "stream_server.event"
-                and response.get("event", {}).get("name", "") == "thinking")
+        return (
+            response.get("object", "") == "stream_server.event"
+            and response.get("event", {}).get("name", "") == "thinking"
+        )
 
     def _set_thinking(self, request: LlmRequest, http_options: dict):
         """Set thinking parameters from request config."""
@@ -654,11 +692,15 @@ def _set_thinking(self, request: LlmRequest, http_options: dict):
                 if thinking_config.thinking_budget <= max_output_tokens:
                     processed_extra_body[const.THINKING_TOKENS] = thinking_config.thinking_budget
                 else:
-                    raise ValueError(f"thinking_budget: {thinking_config.thinking_budget} "
-                                     f"must be between 1024 and {max_output_tokens}")
+                    raise ValueError(
+                        f"thinking_budget: {thinking_config.thinking_budget} "
+                        f"must be between 1024 and {max_output_tokens}"
+                    )
             else:
-                raise ValueError(f"Invalid thinking_budget value: {thinking_config.thinking_budget}. "
-                                 "Must be 0 (disabled), -1 (automatic), or positive integer.")
+                raise ValueError(
+                    f"Invalid thinking_budget value: {thinking_config.thinking_budget}. "
+                    "Must be 0 (disabled), -1 (automatic), or positive integer."
+                )
 
     def _get_thinking_state(self, response: dict) -> int:
         """Get the thinking state from a thinking event.
@@ -687,15 +729,14 @@ def _process_tool_call_delta(self, tool_call_delta: dict, accumulated_tool_calls
 
         # Ensure we have enough slots in accumulated_tool_calls
         while len(accumulated_tool_calls) <= index:
-            accumulated_tool_calls.append({
-                ToolKey.ID: "",
-                ToolKey.TYPE: ToolKey.FUNCTION,
-                ToolKey.FUNCTION: {
-                    ToolKey.NAME: "",
-                    ToolKey.ARGUMENTS: ""
-                },
-                ToolKey.THOUGHT_SIGNATURE: "",
-            })
+            accumulated_tool_calls.append(
+                {
+                    ToolKey.ID: "",
+                    ToolKey.TYPE: ToolKey.FUNCTION,
+                    ToolKey.FUNCTION: {ToolKey.NAME: "", ToolKey.ARGUMENTS: ""},
+                    ToolKey.THOUGHT_SIGNATURE: "",
+                }
+            )
 
         # Capture thought_signature from delta or provider_specific_fields for next-round pass-through
         thought_sig = tool_call_delta.get(ToolKey.THOUGHT_SIGNATURE)
@@ -715,8 +756,11 @@ def _process_tool_call_delta(self, tool_call_delta: dict, accumulated_tool_calls
 
         if ToolKey.FUNCTION in tool_call_delta:
             function_delta = tool_call_delta[ToolKey.FUNCTION]
-            if (ToolKey.NAME in function_delta and function_delta[ToolKey.NAME] is not None
-                    and function_delta[ToolKey.NAME] != ""):
+            if (
+                ToolKey.NAME in function_delta
+                and function_delta[ToolKey.NAME] is not None
+                and function_delta[ToolKey.NAME] != ""
+            ):
                 accumulated_tool_calls[index][ToolKey.FUNCTION][ToolKey.NAME] = function_delta[ToolKey.NAME]
             # If function_delta[ToolKey.NAME] is None or empty, keep the existing name value
             if ToolKey.ARGUMENTS in function_delta and function_delta[ToolKey.ARGUMENTS] is not None:
@@ -729,14 +773,16 @@ def _build_usage_metadata(usage_data: dict) -> GenerateContentResponseUsageMetad
         ``cache_read_input_tokens`` prefers Anthropic/LiteLLM-style top-level fields;
         falls back to OpenAI-style ``prompt_tokens_details.cached_tokens``.
         """
-        completion_details = usage_data.get("completion_tokens_details") or {}
+        completion_details = (
+            usage_data.get("completion_tokens_details") or usage_data.get("output_tokens_details") or {}
+        )
         cache_read = usage_data.get("cache_read_input_tokens")
         if cache_read is None:
-            details = usage_data.get("prompt_tokens_details")
+            details = usage_data.get("prompt_tokens_details") or usage_data.get("input_tokens_details")
             cache_read = details.get("cached_tokens") if isinstance(details, dict) else None
         return GenerateContentResponseUsageMetadata(
-            prompt_token_count=usage_data.get("prompt_tokens", 0),
-            candidates_token_count=usage_data.get("completion_tokens", 0),
+            prompt_token_count=usage_data.get("prompt_tokens", usage_data.get("input_tokens", 0)),
+            candidates_token_count=usage_data.get("completion_tokens", usage_data.get("output_tokens", 0)),
             thoughts_token_count=completion_details.get("reasoning_tokens"),
             total_token_count=usage_data.get("total_tokens", 0),
             cache_read_input_tokens=cache_read,
@@ -849,15 +895,20 @@ def _create_complete_tool_calls(self, accumulated_tool_calls: list[dict]) -> Opt
                         logger.warning("Generated fallback ID '%s' for tool call with missing ID", tool_call_id)
 
                     thought_sig = tool_call_data.get(ToolKey.THOUGHT_SIGNATURE) or None
-                    logger.debug("Creating tool call: id=%s, name=%s, arguments=%s", tool_call_id,
-                                 function_map[ToolKey.NAME], arguments)
+                    logger.debug(
+                        "Creating tool call: id=%s, name=%s, arguments=%s",
+                        tool_call_id,
+                        function_map[ToolKey.NAME],
+                        arguments,
+                    )
                     complete_tool_calls.append(
                         ToolCall(
                             id=tool_call_id,
                             name=function_map[ToolKey.NAME],
                             arguments=arguments,
                             thought_signature=thought_sig,
-                        ))
+                        )
+                    )
                 except json.JSONDecodeError as ex:
                     # Arguments not complete yet, skip this tool call
                     logger.debug("JSON decode error for tool call %s: %s", i, ex)
@@ -996,7 +1047,8 @@ def _process_tool_calls_from_message(self, message: dict) -> Optional[List[ToolC
                         name=tool_call[ToolKey.FUNCTION][ToolKey.NAME],
                         arguments=arguments,
                         thought_signature=thought_sig,
-                    ))
+                    )
+                )
             except (KeyError, json.JSONDecodeError, TypeError) as ex:
                 logger.warning("Failed to parse tool call: %s, error: %s", tool_call, ex)
                 continue
@@ -1051,7 +1103,8 @@ def _create_response_with_content(self, response_dict: dict) -> LlmResponse:
                         tool_call = ToolCall(
                             id=f"call_{uuid.uuid4().hex[:24]}",
                             name=func_call.name,  # type: ignore
-                            arguments=func_call.args)  # type: ignore
+                            arguments=func_call.args,
+                        )  # type: ignore
                         tool_calls.append(tool_call)
             except Exception as ex:  # pylint: disable=broad-except
                 logger.warning("Failed to parse function calls from text content: %s", ex)
@@ -1102,11 +1155,203 @@ def _create_response_with_content(self, response_dict: dict) -> LlmResponse:
 
         return LlmResponse(content=content, usage_metadata=usage, error_code=error_code, response_id=response_id)
 
-    async def _generate_single(self,
-                               api_params: Dict,
-                               request: LlmRequest,
-                               http_options: Dict[str, Any] | None = None,
-                               ctx: InvocationContext | None = None) -> LlmResponse:
+    @staticmethod
+    def _model_dump(value: Any) -> dict:
+        """Return a dictionary for OpenAI SDK models and test doubles."""
+        if isinstance(value, dict):
+            return value
+        return value.model_dump()
+
+    def _convert_messages_to_responses_input(self, messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        """Convert Chat Completions messages to Responses input items."""
+        input_items: List[Dict[str, Any]] = []
+        for message in messages:
+            role = message.get(const.ROLE, const.USER)
+            if role == const.TOOL:
+                input_items.append(
+                    {
+                        "type": "function_call_output",
+                        "call_id": message.get(const.TOOL_CALL_ID, "unknown"),
+                        "output": str(message.get(const.CONTENT, "")),
+                    }
+                )
+                continue
+
+            input_items.extend(message.get(_RESPONSES_INPUT_ITEMS) or [])
+
+            content = message.get(const.CONTENT)
+            if content not in (None, "", []):
+                if isinstance(content, list):
+                    converted_content = []
+                    for item in content:
+                        item_type = item.get("type")
+                        if item_type == "image_url":
+                            image = item.get("image_url") or {}
+                            converted_content.append(
+                                {
+                                    "type": "input_image",
+                                    "image_url": image.get("url", ""),
+                                    "detail": image.get("detail", "auto"),
+                                }
+                            )
+                        elif item_type == "text":
+                            converted_content.append(
+                                {
+                                    "type": "output_text" if role == const.ASSISTANT else "input_text",
+                                    "text": item.get("text", ""),
+                                }
+                            )
+                        else:
+                            converted_content.append(item)
+                    content = converted_content
+                input_items.append({"role": role, "content": content})
+
+            for tool_call in message.get(const.TOOL_CALLS, []) or []:
+                function = tool_call.get(ToolKey.FUNCTION, {})
+                input_items.append(
+                    {
+                        "type": "function_call",
+                        "call_id": tool_call.get(ToolKey.ID) or f"call_{uuid.uuid4().hex[:24]}",
+                        "name": function.get(ToolKey.NAME, ""),
+                        "arguments": function.get(ToolKey.ARGUMENTS, "{}"),
+                    }
+                )
+        return input_items
+
+    @staticmethod
+    def _convert_tools_to_responses_format(tools: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        """Flatten Chat Completions function definitions for Responses."""
+        converted = []
+        for tool in tools:
+            if tool.get("type") != "function":
+                converted.append(tool)
+                continue
+            function = dict(tool.get("function") or {})
+            function["type"] = "function"
+            converted.append(function)
+        return converted
+
+    @staticmethod
+    def _convert_response_format_to_responses(response_format: Dict[str, Any]) -> Dict[str, Any]:
+        """Convert Chat Completions response_format to Responses text config."""
+        if response_format.get("type") == "json_schema":
+            schema = dict(response_format.get("json_schema") or {})
+            return {"format": {"type": "json_schema", **schema}}
+        return {"format": dict(response_format)}
+
+    def _convert_api_params_to_responses(self, api_params: Dict[str, Any]) -> Dict[str, Any]:
+        """Translate shared generation parameters to ``responses.create``."""
+        responses_params: Dict[str, Any] = {
+            "model": api_params[ApiParamsKey.MODEL],
+            "input": self._convert_messages_to_responses_input(api_params[ApiParamsKey.MESSAGES]),
+            "stream": api_params[ApiParamsKey.STREAM],
+        }
+        parameter_map = {
+            ApiParamsKey.TEMPERATURE: "temperature",
+            ApiParamsKey.TOP_P: "top_p",
+            ApiParamsKey.TOOL_CHOICE: "tool_choice",
+            ApiParamsKey.PARALLEL_TOOL_CALLS: "parallel_tool_calls",
+            ApiParamsKey.PROMPT_CACHE_KEY: "prompt_cache_key",
+            ApiParamsKey.PROMPT_CACHE_RETENTION: "prompt_cache_retention",
+            ApiParamsKey.TOP_LOGPROBS: "top_logprobs",
+        }
+        for source, target in parameter_map.items():
+            if source in api_params:
+                responses_params[target] = api_params[source]
+
+        max_output_tokens = api_params.get(ApiParamsKey.MAX_COMPLETION_TOKENS) or api_params.get(
+            ApiParamsKey.MAX_TOKENS
+        )
+        if max_output_tokens:
+            responses_params["max_output_tokens"] = max_output_tokens
+        if ApiParamsKey.TOOLS in api_params:
+            responses_params["tools"] = self._convert_tools_to_responses_format(api_params[ApiParamsKey.TOOLS])
+        if ApiParamsKey.RESPONSE_FORMAT in api_params:
+            responses_params["text"] = self._convert_response_format_to_responses(
+                api_params[ApiParamsKey.RESPONSE_FORMAT]
+            )
+
+        responses_params.update(self.responses_api_params)
+        if responses_params.get("store") is False:
+            include = list(responses_params.get("include") or [])
+            if "reasoning.encrypted_content" not in include:
+                include.append("reasoning.encrypted_content")
+            responses_params["include"] = include
+        return responses_params
+
+    @staticmethod
+    def _responses_error(response_dict: dict) -> tuple[Optional[str], Optional[str]]:
+        status = response_dict.get("status")
+        if status not in {"failed", "incomplete", "cancelled"}:
+            return None, None
+        error = response_dict.get("error") or response_dict.get("incomplete_details") or {}
+        if isinstance(error, dict):
+            return str(error.get("code") or status), str(error.get("message") or error.get("reason") or status)
+        return str(status), str(error)
+
+    def _create_responses_response(self, response_dict: dict) -> LlmResponse:
+        """Convert a completed Responses payload into ``LlmResponse``."""
+        parts = []
+        for item in response_dict.get("output") or []:
+            item_type = item.get("type")
+            if item_type == "reasoning":
+                summaries = item.get("summary") or []
+                if not summaries:
+                    part = Part.from_text(text="")
+                    part.thought = True
+                    part.thought_signature = json.dumps(item, ensure_ascii=False).encode("utf-8")
+                    parts.append(part)
+                for index, summary in enumerate(summaries):
+                    text = summary.get("text") if isinstance(summary, dict) else None
+                    if text:
+                        part = Part.from_text(text=text)
+                        part.thought = True
+                        if index == 0:
+                            part.thought_signature = json.dumps(item, ensure_ascii=False).encode("utf-8")
+                        parts.append(part)
+            elif item_type == "message":
+                for output_part in item.get("content") or []:
+                    text = output_part.get("text") or output_part.get("refusal")
+                    if text:
+                        part = Part.from_text(text=text)
+                        part.thought = False
+                        parts.append(part)
+            elif item_type == "function_call":
+                arguments = json_loads_repair(item.get("arguments") or "{}")
+                if not isinstance(arguments, dict):
+                    logger.warning("Skipping Responses function call with non-dict arguments: %r", arguments)
+                    continue
+                part = Part.from_function_call(name=item.get("name", ""), args=arguments)
+                part.function_call.id = item.get("call_id") or item.get("id")  # type: ignore
+                parts.append(part)
+
+        content = Content(parts=parts, role=const.MODEL) if parts else None
+        usage = self._process_usage_from_response(response_dict)
+        error_code, error_message = self._responses_error(response_dict)
+        return LlmResponse(
+            content=content,
+            usage_metadata=usage,
+            error_code=error_code,
+            error_message=error_message,
+            response_id=response_dict.get("id"),
+        )
+
+    async def _generate_responses_single(self, api_params: Dict[str, Any]) -> LlmResponse:
+        """Generate one non-streaming response through the Responses API."""
+        client = self._create_async_client()
+        try:
+            response = await client.responses.create(**api_params)
+            return self._create_responses_response(self._model_dump(response))
+        finally:
+            await self._http_client_provider.close_http_client(client)
+
+    async def _generate_single(
+        self,
+        api_params: Dict,
+        request: LlmRequest,
+        http_options: Dict[str, Any] | None = None,
+        ctx: InvocationContext | None = None,
+    ) -> LlmResponse:
         """Generate a single response (non-streaming)."""
         if http_options is None:
             http_options = {}
@@ -1157,7 +1402,8 @@ def _convert_tools_to_openai_format(self, tools: List[Tool]) -> List[Dict[str, A
                     # Convert parameters schema - always include parameters field
                     if func_decl.parameters:
                         openai_tool["function"]["parameters"] = self._convert_schema_to_openai_format(
-                            func_decl.parameters)
+                            func_decl.parameters
+                        )
                     else:
                         # When parameters are empty, provide the proper OpenAI format structure
                         openai_tool["function"]["parameters"] = {"type": "object", "properties": {}}
@@ -1295,22 +1541,14 @@ def _build_response_format(self, config: GenerateContentConfig) -> Optional[Dict
                 openai_schema = self._ensure_additional_properties_false(openai_schema)
                 return {
                     "type": "json_schema",
-                    "json_schema": {
-                        "name": "response_schema",
-                        "schema": openai_schema,
-                        "strict": True
-                    },
+                    "json_schema": {"name": "response_schema", "schema": openai_schema, "strict": True},
                 }
             elif config.response_json_schema:
                 # Use provided JSON schema directly
                 processed_schema = self._ensure_additional_properties_false(config.response_json_schema)
                 return {
                     "type": "json_schema",
-                    "json_schema": {
-                        "name": "response_schema",
-                        "schema": processed_schema,
-                        "strict": True
-                    },
+                    "json_schema": {"name": "response_schema", "schema": processed_schema, "strict": True},
                 }
             else:
                 # Basic JSON mode
@@ -1428,13 +1666,13 @@ def _log_unsupported_config_options(self, config: GenerateContentConfig) -> None
         if unsupported_options:
             logger.warning(
                 "The following configuration options are not supported in OpenAI models and will be ignored: %s",
-                ', '.join(unsupported_options))
+                ", ".join(unsupported_options),
+            )
 
     @override
-    async def _generate_async_impl(self,
-                                   request: LlmRequest,
-                                   stream: bool = False,
-                                   ctx: InvocationContext | None = None) -> AsyncGenerator[LlmResponse, None]:
+    async def _generate_async_impl(
+        self, request: LlmRequest, stream: bool = False, ctx: InvocationContext | None = None
+    ) -> AsyncGenerator[LlmResponse, None]:
         """Generate content asynchronously."""
         self.validate_request(request)
 
@@ -1443,8 +1681,12 @@ async def _generate_async_impl(self,
 
         # Update request with merged config
         request.config = merged_config
-        if (request.config and request.config.tools and self._adapter.requires_add_tools_to_prompt()
-                and not self.add_tools_to_prompt):
+        if (
+            request.config
+            and request.config.tools
+            and self._adapter.requires_add_tools_to_prompt()
+            and not self.add_tools_to_prompt
+        ):
             raise ValueError(f"{self._model_name} requires add_tools_to_prompt=True when tools are used.")
 
         # Prepare OpenAI API parameters
@@ -1481,18 +1723,23 @@ async def _generate_async_impl(self,
                     api_params[ApiParamsKey.STOP] = request.config.stop_sequences
 
                 # Additional OpenAI-specific parameters
-                if (request.config.frequency_penalty is not None
-                        and not self._adapter.should_skip_config_param("frequency_penalty")):
+                if request.config.frequency_penalty is not None and not self._adapter.should_skip_config_param(
+                    "frequency_penalty"
+                ):
                     api_params[ApiParamsKey.FREQUENCY_PENALTY] = request.config.frequency_penalty
-                if (request.config.presence_penalty is not None
-                        and not self._adapter.should_skip_config_param("presence_penalty")):
+                if request.config.presence_penalty is not None and not self._adapter.should_skip_config_param(
+                    "presence_penalty"
+                ):
                     api_params[ApiParamsKey.PRESENCE_PENALTY] = request.config.presence_penalty
                 if request.config.seed is not None and not self._adapter.should_skip_config_param("seed"):
                     api_params[ApiParamsKey.SEED] = request.config.seed
 
                 # Handle candidate count (maps to OpenAI's 'n' parameter)
-                if (request.config.candidate_count is not None and request.config.candidate_count > 0
-                        and not self._adapter.should_skip_config_param("candidate_count")):
+                if (
+                    request.config.candidate_count is not None
+                    and request.config.candidate_count > 0
+                    and not self._adapter.should_skip_config_param("candidate_count")
+                ):
                     api_params[ApiParamsKey.N] = request.config.candidate_count
 
                 # Handle logprobs configuration
@@ -1528,21 +1775,207 @@ async def _generate_async_impl(self,
         http_options = {}
         if request.config:
             http_options = self._extract_http_options(request.config)
-        # set thinking params
-        self._set_thinking(request, http_options)
+        if self.use_responses_api:
+            api_params = self._convert_api_params_to_responses(api_params)
+            if (
+                request.config
+                and request.config.thinking_config
+                and request.config.thinking_config.include_thoughts
+                and request.config.thinking_config.thinking_budget != 0
+            ):
+                reasoning = dict(api_params.get("reasoning") or {})
+                reasoning.setdefault("summary", "auto")
+                api_params["reasoning"] = reasoning
+        else:
+            # Chat Completions provider-specific thinking params.
+            self._set_thinking(request, http_options)
 
         if stream:
-            async for response in self._generate_stream(api_params, request, http_options, ctx):
+            generator = (
+                self._generate_responses_stream(api_params, request, http_options)
+                if self.use_responses_api
+                else self._generate_stream(api_params, request, http_options, ctx)
+            )
+            async for response in generator:
                 yield response
         else:
-            response = await self._generate_single(api_params, request, http_options, ctx)
+            response = (
+                await self._generate_responses_single({**api_params, **http_options})
+                if self.use_responses_api
+                else await self._generate_single(api_params, request, http_options, ctx)
+            )
             yield response
 
-    async def _generate_stream(self,
-                               api_params: Dict,
-                               request: LlmRequest,
-                               http_options: Dict[str, Any] | None = None,
-                               ctx: InvocationContext | None = None) -> AsyncGenerator[LlmResponse, None]:
+    async def _generate_responses_stream(
+        self,
+        api_params: Dict[str, Any],
+        request: LlmRequest,
+        http_options: Optional[Dict[str, Any]] = None,
+    ) -> AsyncGenerator[LlmResponse, None]:
+        """Generate streaming responses through ``responses.create``."""
+        client = self._create_async_client()
+        response_id: Optional[str] = None
+        completed_response: Optional[dict] = None
+        accumulated_text = ""
+        accumulated_reasoning = ""
+        function_calls: Dict[str, Dict[str, Any]] = {}
+        function_order: List[str] = []
+        streaming_tool_names = getattr(request, "streaming_tool_names", None) or set()
+
+        def upsert_function(item: dict) -> tuple[str, Dict[str, Any]]:
+            item_id = str(item.get("id") or item.get("call_id") or f"fc_{len(function_order)}")
+            if item_id not in function_calls:
+                function_calls[item_id] = {
+                    "type": "function_call",
+                    "id": item.get("id"),
+                    "call_id": item.get("call_id") or item.get("id"),
+                    "name": item.get("name", ""),
+                    "arguments": item.get("arguments", ""),
+                }
+                function_order.append(item_id)
+            else:
+                current = function_calls[item_id]
+                for key in ("id", "call_id", "name"):
+                    if item.get(key):
+                        current[key] = item[key]
+                if item.get("arguments"):
+                    current["arguments"] = item["arguments"]
+            return item_id, function_calls[item_id]
+
+        try:
+            response = await client.responses.create(**api_params, **(http_options or {}))
+            if response is None:
+                raise ValueError("Empty response from Responses API")
+
+            async for event in response:
+                event_dict = self._model_dump(event)
+                event_type = event_dict.get("type", "")
+                logger.debug("OpenAI Responses event: %s", json.dumps(event_dict, ensure_ascii=False))
+
+                response_data = event_dict.get("response") or {}
+                if response_id is None and response_data.get("id"):
+                    response_id = response_data["id"]
+                if response_id is None and event_dict.get("response_id"):
+                    response_id = event_dict["response_id"]
+
+                if event_type == "error":
+                    completed_response = {
+                        "id": response_id,
+                        "status": "failed",
+                        "error": {
+                            "code": event_dict.get("code") or "responses_stream_error",
+                            "message": event_dict.get("message") or "Responses stream failed",
+                        },
+                        "output": [],
+                    }
+                    continue
+
+                if event_type in {"response.output_text.delta", "response.refusal.delta"}:
+                    delta = event_dict.get("delta") or ""
+                    if delta:
+                        accumulated_text += delta
+                        part = Part.from_text(text=delta)
+                        part.thought = False
+                        yield LlmResponse(
+                            content=Content(parts=[part], role=const.MODEL),
+                            partial=True,
+                            response_id=response_id,
+                            custom_metadata={const.CHUNK: event_dict},
+                        )
+                    continue
+
+                if event_type in {"response.reasoning_summary_text.delta", "response.reasoning_text.delta"}:
+                    delta = event_dict.get("delta") or ""
+                    if delta:
+                        accumulated_reasoning += delta
+                        part = Part.from_text(text=delta)
+                        part.thought = True
+                        yield LlmResponse(
+                            content=Content(parts=[part], role=const.MODEL),
+                            partial=True,
+                            response_id=response_id,
+                            custom_metadata={const.CHUNK: event_dict},
+                        )
+                    continue
+
+                if event_type in {"response.output_item.added", "response.output_item.done"}:
+                    item = event_dict.get("item") or {}
+                    if item.get("type") == "function_call":
+                        upsert_function(item)
+                    continue
+
+                if event_type == "response.function_call_arguments.delta":
+                    item_id = str(event_dict.get("item_id") or f"fc_{event_dict.get('output_index', 0)}")
+                    if item_id not in function_calls:
+                        function_order.append(item_id)
+                        function_calls[item_id] = {
+                            "type": "function_call",
+                            "id": event_dict.get("item_id"),
+                            "call_id": event_dict.get("call_id") or event_dict.get("item_id"),
+                            "name": event_dict.get("name", ""),
+                            "arguments": "",
+                        }
+                    function_call = function_calls[item_id]
+                    delta = event_dict.get("delta") or ""
+                    function_call["arguments"] += delta
+                    name = function_call.get("name", "")
+                    if delta and name and name in streaming_tool_names:
+                        part = Part.from_function_call(name=name, args={const.TOOL_STREAMING_ARGS: delta})
+                        part.function_call.id = function_call.get("call_id")  # type: ignore
+                        yield LlmResponse(
+                            content=Content(parts=[part], role=const.MODEL),
+                            partial=True,
+                            response_id=response_id,
+                            custom_metadata={const.CHUNK: event_dict, const.TOOL_STREAMING: True},
+                        )
+                    continue
+
+                if event_type == "response.function_call_arguments.done":
+                    item_id = str(event_dict.get("item_id") or f"fc_{event_dict.get('output_index', 0)}")
+                    item = event_dict.get("item") or {}
+                    if item.get("type") == "function_call":
+                        upsert_function(item)
+                    elif item_id in function_calls:
+                        function_calls[item_id]["arguments"] = event_dict.get("arguments") or ""
+                    continue
+
+                if event_type in {"response.completed", "response.failed", "response.incomplete"}:
+                    completed_response = response_data
+
+            if completed_response is None:
+                output = []
+                if accumulated_reasoning:
+                    output.append(
+                        {
+                            "type": "reasoning",
+                            "summary": [{"type": "summary_text", "text": accumulated_reasoning}],
+                        }
+                    )
+                if accumulated_text:
+                    output.append(
+                        {
+                            "type": "message",
+                            "role": "assistant",
+                            "content": [{"type": "output_text", "text": accumulated_text}],
+                        }
+                    )
+                output.extend(function_calls[item_id] for item_id in function_order)
+                completed_response = {"id": response_id, "status": "completed", "output": output}
+
+            final_response = self._create_responses_response(completed_response)
+            final_response.partial = False
+            final_response.custom_metadata = {"stream_complete": True}
+            yield final_response
+        finally:
+            await self._http_client_provider.close_http_client(client)
+
+    async def _generate_stream(
+        self,
+        api_params: Dict,
+        request: LlmRequest,
+        http_options: Dict[str, Any] | None = None,
+        ctx: InvocationContext | None = None,
+    ) -> AsyncGenerator[LlmResponse, None]:
         """Generate streaming responses."""
         if http_options is None:
             http_options = {}
@@ -1555,7 +1988,7 @@ async def _generate_stream(self,
         response_id: str | None = None  # Track response ID from API
 
         # For streaming tool call arguments - get the set of tool names that should stream
-        streaming_tool_names = getattr(request, 'streaming_tool_names', None) or set()
+        streaming_tool_names = getattr(request, "streaming_tool_names", None) or set()
 
         # Create tool prompt instance for streaming if needed
         tool_prompt = None
@@ -1613,9 +2046,9 @@ async def _generate_stream(self,
                     if streaming_tool_names and delta_arguments and accumulated_tool_calls:
                         # Yield streaming tool call event with delta arguments
                         # Only for tools in streaming_tool_names
-                        streaming_event = self._create_streaming_tool_call_response(accumulated_tool_calls,
-                                                                                    delta_arguments,
-                                                                                    streaming_tool_names)
+                        streaming_event = self._create_streaming_tool_call_response(
+                            accumulated_tool_calls, delta_arguments, streaming_tool_names
+                        )
                         if streaming_event:
                             yield streaming_event
 
@@ -1633,11 +2066,15 @@ async def _generate_stream(self,
                     reasoning_content = delta.get(const.REASONING_CONTENT)
                     if reasoning_content is not None:
                         partial_text = reasoning_content
-                        if (tool_prompt and streaming_text_filter_state is not None
-                                and self._adapter.should_filter_reasoning_text()):
+                        if (
+                            tool_prompt
+                            and streaming_text_filter_state is not None
+                            and self._adapter.should_filter_reasoning_text()
+                        ):
                             reasoning_filter_state = streaming_text_filter_state["reasoning"]
-                            partial_text = self._adapter.filter_streaming_text(reasoning_content,
-                                                                               reasoning_filter_state)
+                            partial_text = self._adapter.filter_streaming_text(
+                                reasoning_content, reasoning_filter_state
+                            )
                         if not partial_text:
                             continue
 
@@ -1649,10 +2086,12 @@ async def _generate_stream(self,
                         content_part.thought = True
 
                         partial_content = Content(parts=[content_part], role=const.MODEL)
-                        yield LlmResponse(content=partial_content,
-                                          partial=True,
-                                          response_id=response_id,
-                                          custom_metadata={const.CHUNK: chunk_dict})
+                        yield LlmResponse(
+                            content=partial_content,
+                            partial=True,
+                            response_id=response_id,
+                            custom_metadata={const.CHUNK: chunk_dict},
+                        )
 
                 # Handle regular content
                 if delta.get(const.CONTENT):
@@ -1677,10 +2116,12 @@ async def _generate_stream(self,
                         content_part.thought = is_thinking
 
                         partial_content = Content(parts=[content_part], role=const.MODEL)
-                        yield LlmResponse(content=partial_content,
-                                          partial=True,
-                                          response_id=response_id,
-                                          custom_metadata={const.CHUNK: chunk_dict})
+                        yield LlmResponse(
+                            content=partial_content,
+                            partial=True,
+                            response_id=response_id,
+                            custom_metadata={const.CHUNK: chunk_dict},
+                        )
 
                 # Handle usage
                 usage = self._process_usage(chunk_dict)
@@ -1690,16 +2131,19 @@ async def _generate_stream(self,
             if tool_prompt and streaming_text_filter_state is not None:
                 if self._adapter.should_filter_reasoning_text():
                     flushed_reasoning_text = self._adapter.flush_streaming_text(
-                        streaming_text_filter_state["reasoning"])
+                        streaming_text_filter_state["reasoning"]
+                    )
                     if flushed_reasoning_text:
                         thought_content += flushed_reasoning_text
                         content_part = Part.from_text(text=flushed_reasoning_text)
                         content_part.thought = True
                         partial_content = Content(parts=[content_part], role=const.MODEL)
-                        yield LlmResponse(content=partial_content,
-                                          partial=True,
-                                          response_id=response_id,
-                                          custom_metadata={"stream_filter_flushed": "reasoning"})
+                        yield LlmResponse(
+                            content=partial_content,
+                            partial=True,
+                            response_id=response_id,
+                            custom_metadata={"stream_filter_flushed": "reasoning"},
+                        )
 
                 flushed_content_text = self._adapter.flush_streaming_text(streaming_text_filter_state["content"])
                 if flushed_content_text:
@@ -1708,10 +2152,12 @@ async def _generate_stream(self,
                     content_part = Part.from_text(text=flushed_content_text)
                     content_part.thought = is_thinking
                     partial_content = Content(parts=[content_part], role=const.MODEL)
-                    yield LlmResponse(content=partial_content,
-                                      partial=True,
-                                      response_id=response_id,
-                                      custom_metadata={"stream_filter_flushed": "content"})
+                    yield LlmResponse(
+                        content=partial_content,
+                        partial=True,
+                        response_id=response_id,
+                        custom_metadata={"stream_filter_flushed": "content"},
+                    )
 
             # Yield final complete response
             final_content = None
@@ -1729,17 +2175,19 @@ async def _generate_stream(self,
             if tool_prompt and accumulated_content and not complete_tool_calls:
                 try:
                     parsed_function_calls = self._adapter.parse_tool_prompt_function_calls(
-                        accumulated_content, tool_prompt)
+                        accumulated_content, tool_prompt
+                    )
                     if parsed_function_calls:
                         # Convert FunctionCall objects to ToolCall objects
                         complete_tool_calls = []
                         for func_call in parsed_function_calls:
-                            tool_call = ToolCall(id=f"call_{uuid.uuid4().hex[:24]}",
-                                                 name=func_call.name,
-                                                 arguments=func_call.args)
+                            tool_call = ToolCall(
+                                id=f"call_{uuid.uuid4().hex[:24]}", name=func_call.name, arguments=func_call.args
+                            )
                             complete_tool_calls.append(tool_call)
-                        logger.debug("Parsed %s function calls from final accumulated content",
-                                     len(complete_tool_calls))
+                        logger.debug(
+                            "Parsed %s function calls from final accumulated content", len(complete_tool_calls)
+                        )
                 except Exception as ex:  # pylint: disable=broad-except
                     logger.warning("Failed to parse function calls from final accumulated content: %s", ex)
 
diff --git a/trpc_agent_sdk/tools/__init__.py b/trpc_agent_sdk/tools/__init__.py
index 715da600..5ad6788e 100644
--- a/trpc_agent_sdk/tools/__init__.py
+++ b/trpc_agent_sdk/tools/__init__.py
@@ -15,93 +15,79 @@
     from trpc_agent_sdk.agents.sub_agent import DynamicSubAgentTool as DynamicSubAgentTool  # noqa: F401
     from trpc_agent_sdk.agents.sub_agent import SpawnSubAgentTool as SpawnSubAgentTool  # noqa: F401
 
-from ._agent_tool import AGENT_TOOL_APP_NAME_SUFFIX
-from ._agent_tool import AgentTool
+from ._agent_tool import AGENT_TOOL_APP_NAME_SUFFIX, AgentTool
 from ._base_tool import BaseTool
 from ._constants import TOOL_NAME
-from ._context_var import get_tool_var
-from ._context_var import reset_tool_var
-from ._context_var import set_tool_var
+from ._context_var import get_tool_var, reset_tool_var, set_tool_var
 from ._default_toolset import DefaultToolSet
 from ._function_tool import FunctionTool
-from ._load_memory_tool import LoadMemoryResponse
-from ._load_memory_tool import LoadMemoryTool
-from ._load_memory_tool import load_memory
-from ._load_memory_tool import load_memory_tool
-from ._long_running_tool import LongRunningFunctionTool
-from ._preload_memory_tool import PreloadMemoryTool
-from ._preload_memory_tool import preload_memory_tool
-from ._registry import ToolRegistry
-from ._registry import ToolSetRegistry
-from ._registry import ToolType
-from ._registry import get_tool
-from ._registry import get_tool_set
-from ._registry import register_tool
-from ._registry import register_tool_set
+from ._load_memory_tool import LoadMemoryResponse, LoadMemoryTool, load_memory, load_memory_tool
+from ._long_running_tool import LongRunningFunctionTool, is_tool_execution_error
+from ._preload_memory_tool import PreloadMemoryTool, preload_memory_tool
+from ._registry import ToolRegistry, ToolSetRegistry, ToolType, get_tool, get_tool_set, register_tool, register_tool_set
 from ._set_model_response_tool import SetModelResponseTool
 from ._streaming_function_tool import StreamingFunctionTool
 from ._streaming_progress_tool import StreamingProgressTool
-from ._todo_tool import DEFAULT_NUDGE_MESSAGE
-from ._todo_tool import DEFAULT_STATE_KEY_PREFIX
-from ._todo_tool import DEFAULT_TODO_DESCRIPTION
-from ._todo_tool import DEFAULT_TODO_PROMPT
-from ._todo_tool import TodoItem
-from ._todo_tool import TodoStatus
-from ._todo_tool import TodoWriteTool
-from ._todo_tool import get_todos
-from ._todo_tool import render_todos
-from ._todo_tool import state_key
-from ._todo_tool import validate_todos
-from ._tool_adapter import convert_toolunion_to_tool_list
-from ._tool_adapter import create_tool
-from ._tool_adapter import create_toolset
-from .task_tools import DEFAULT_TASK_PROMPT
-from .task_tools import TaskCreateTool
-from .task_tools import TaskGetTool
-from .task_tools import TaskListSummary
-from .task_tools import TaskListTool
-from .task_tools import TaskRecord
-from .task_tools import TaskStatus
-from .task_tools import TaskStore
-from .task_tools import TaskToolSet
-from .task_tools import TaskUpdateTool
-from .task_tools import get_task_store
-from .task_tools import render_task_list
-from .goal_tools import GoalOptions
-from .goal_tools import GoalRecord
-from .goal_tools import GoalStatus
-from .goal_tools import GoalToolSet
-from .goal_tools import OnRetry
-from .goal_tools import RetryEvent
-from .goal_tools import get_goal_record
-from .goal_tools import setup_goal
-from .goal_tools import render_goal
+from ._todo_tool import (
+    DEFAULT_NUDGE_MESSAGE,
+    DEFAULT_STATE_KEY_PREFIX,
+    DEFAULT_TODO_DESCRIPTION,
+    DEFAULT_TODO_PROMPT,
+    TodoItem,
+    TodoStatus,
+    TodoWriteTool,
+    get_todos,
+    render_todos,
+    state_key,
+    validate_todos,
+)
+from ._tool_adapter import convert_toolunion_to_tool_list, create_tool, create_toolset
 from ._transfer_to_agent_tool import transfer_to_agent
-from ._webfetch_tool import FetchResult
-from ._webfetch_tool import WebFetchTool
-from ._websearch_tool import SearchHit
-from ._websearch_tool import WebSearchResult
-from ._websearch_tool import WebSearchTool
-from .file_tools import BashTool
-from .file_tools import EditTool
-from .file_tools import FileToolSet
-from .file_tools import GlobTool
-from .file_tools import GrepTool
-from .file_tools import ReadTool
-from .file_tools import WriteTool
-from .mcp_tool import MCPTool
-from .mcp_tool import MCPToolset
-from .mcp_tool import McpConnectionParamsType
-from .mcp_tool import McpStdioServerParameters
-from .mcp_tool import SseConnectionParams
-from .mcp_tool import StdioConnectionParams
-from .mcp_tool import StreamableHTTPConnectionParams
-from .mcp_tool import patch_mcp_cancel_scope_exit_issue
-from .utils import build_function_declaration
-from .utils import from_function_with_options
-from .utils import get_required_fields
-from .utils import parse_schema_from_parameter
-from .utils import register_checker
+from ._webfetch_tool import FetchResult, WebFetchTool
+from ._websearch_tool import SearchHit, WebSearchResult, WebSearchTool
+from .file_tools import BashTool, EditTool, FileToolSet, GlobTool, GrepTool, ReadTool, WriteTool
+from .goal_tools import (
+    GoalOptions,
+    GoalRecord,
+    GoalStatus,
+    GoalToolSet,
+    OnRetry,
+    RetryEvent,
+    get_goal_record,
+    render_goal,
+    setup_goal,
+)
+from .mcp_tool import (
+    McpConnectionParamsType,
+    McpStdioServerParameters,
+    MCPTool,
+    MCPToolset,
+    SseConnectionParams,
+    StdioConnectionParams,
+    StreamableHTTPConnectionParams,
+    patch_mcp_cancel_scope_exit_issue,
+)
+from .task_tools import (
+    DEFAULT_TASK_PROMPT,
+    TaskCreateTool,
+    TaskGetTool,
+    TaskListSummary,
+    TaskListTool,
+    TaskRecord,
+    TaskStatus,
+    TaskStore,
+    TaskToolSet,
+    TaskUpdateTool,
+    get_task_store,
+    render_task_list,
+)
+from .utils import (
+    build_function_declaration,
+    from_function_with_options,
+    get_required_fields,
+    parse_schema_from_parameter,
+    register_checker,
+)
 
 __all__ = [
     "ToolPredicate",
@@ -119,6 +105,7 @@
     "load_memory",
     "load_memory_tool",
     "LongRunningFunctionTool",
+    "is_tool_execution_error",
     "PreloadMemoryTool",
     "preload_memory_tool",
     "ToolRegistry",
@@ -207,6 +194,7 @@
 def __getattr__(name):
     if name in _LAZY_REEXPORTS:
         import importlib
+
         module_name, attr = _LAZY_REEXPORTS[name]
         obj = getattr(importlib.import_module(module_name), attr)
         globals()[name] = obj  # cache: subsequent accesses skip __getattr__
diff --git a/trpc_agent_sdk/tools/_long_running_tool.py b/trpc_agent_sdk/tools/_long_running_tool.py
index 4e70954f..d6125b90 100644
--- a/trpc_agent_sdk/tools/_long_running_tool.py
+++ b/trpc_agent_sdk/tools/_long_running_tool.py
@@ -7,8 +7,8 @@
 
 from __future__ import annotations
 
-from typing import Callable
-from typing import Optional
+from typing import Callable, Optional
+
 from typing_extensions import override
 
 from trpc_agent_sdk.filter import BaseFilter
@@ -16,6 +16,39 @@
 
 from ._function_tool import FunctionTool
 
+_TOOL_ERROR_MARKERS = (
+    "an error occurred while parsing tool arguments",
+    "json parse error",
+    "invalid tool arguments",
+    "tool arguments validation failed",
+)
+
+
+def is_tool_execution_error(response: object) -> bool:
+    """Return whether a tool response represents invocation failure.
+
+    A LongRunningFunctionTool is promoted to HITL only after the tool call
+    reached the tool boundary. Provider-side argument parsing errors are
+    returned as ordinary function responses by some adapters and must not be
+    mistaken for a successful human-interaction checkpoint.
+    """
+    if not isinstance(response, dict):
+        return False
+
+    status = str(response.get("status") or "").strip().lower()
+    if status in {"error", "failed", "failure", "invalid"}:
+        return True
+    if response.get("error") or response.get("error_message"):
+        return True
+
+    for key in ("result", "message"):
+        value = response.get(key)
+        if isinstance(value, str):
+            normalized = value.strip().lower()
+            if any(marker in normalized for marker in _TOOL_ERROR_MARKERS):
+                return True
+    return False
+
 
 class LongRunningFunctionTool(FunctionTool):
     """A function tool that returns the result asynchronously.
@@ -34,10 +67,9 @@ class LongRunningFunctionTool(FunctionTool):
         is_long_running: Whether the tool is a long running operation.
     """
 
-    def __init__(self,
-                 func: Callable,
-                 filters_name: Optional[list[str]] = None,
-                 filters: Optional[list[BaseFilter]] = None):
+    def __init__(
+        self, func: Callable, filters_name: Optional[list[str]] = None, filters: Optional[list[BaseFilter]] = None
+    ):
         """Initialize the long running function tool.
 
         Args:
@@ -58,9 +90,11 @@ def _get_declaration(self) -> Optional[FunctionDeclaration]:
         """
         declaration = super()._get_declaration()
         if declaration:
-            instruction = ("\n\nNOTE: This is a long-running operation. Do not call this tool"
-                           " again if it has already returned some intermediate or pending"
-                           " status.")
+            instruction = (
+                "\n\nNOTE: This is a long-running operation. Do not call this tool"
+                " again if it has already returned some intermediate or pending"
+                " status."
+            )
             if declaration.description:
                 declaration.description += instruction
             else: