diff --git a/README.md b/README.md
index e57c2fb5..0e693665 100644
--- a/README.md
+++ b/README.md
@@ -594,6 +594,7 @@ uv run python main.py [example] [flag-key] [user-input]
 | --- | --- | --- |
 | `agent` *(default)* | `uv run python main.py agent` | `config()` via the global registry — switches providers without code changes |
 | `graph` | `uv run python main.py graph` | `graph()` multi-agent workflow driven by a LaunchDarkly agent graph flag |
+| `graph-history` | `uv run python main.py graph-history` | `graph().invoke()` with multimodal `history` forwarded to the root node |
 | `openai-only` | `uv run python main.py openai-only` | `config()` with a custom `Registry` restricted to OpenAI handlers |
 | `streaming` | `uv run python main.py streaming` | `config().stream()` — token-by-token output |
 
diff --git a/examples/graph_history.py b/examples/graph_history.py
new file mode 100644
index 00000000..f1485434
--- /dev/null
+++ b/examples/graph_history.py
@@ -0,0 +1,100 @@
+"""
+Example: graph().invoke() with multimodal conversation history.
+
+Passes a `history` list containing an image content block to a graph flag. Only
+the root node receives the history; downstream nodes see it through the normal
+node-to-node data passing. The image is a generated solid red square, so the
+model naming the colour is the signal that the image actually reached the
+provider.
+
+Usage (via main.py):
+    python main.py graph-history <graph-flag-key> "<user input>"
+"""
+
+from __future__ import annotations
+
+import json
+import re
+import sys
+from typing import Any
+
+import examples.register  # noqa: F401 – side-effect: populate global_registry
+from examples.utils import new_context, solid_color_png_base64, write_output
+from launchdarkly_ai_server import global_registry, graph
+
+IMAGE_BLOCK = {
+    "type": "image",
+    "source": {
+        "type": "base64",
+        "media_type": "image/png",
+        "data": solid_color_png_base64((255, 0, 0)),
+    },
+}
+
+COLOR_QUESTION = (
+    "What colour is the square in the image I shared? Answer with just the colour name."
+)
+
+# Two supported shapes: history that carries only context (the user turn arrives
+# as user_input), and history that already ends with the user turn (user_input
+# is empty).
+SCENARIOS: list[dict[str, Any]] = [
+    {
+        "name": "image-in-history + question as user_input",
+        "history": [{"role": "user", "content": [IMAGE_BLOCK]}],
+        "user_input": COLOR_QUESTION,
+    },
+    {
+        "name": "history ends with the user turn, empty user_input",
+        "history": [
+            {"role": "user", "content": "I am going to share an image with you."},
+            {"role": "assistant", "content": "Sure — go ahead and share it."},
+            {
+                "role": "user",
+                "content": [IMAGE_BLOCK, {"type": "text", "text": COLOR_QUESTION}],
+            },
+        ],
+        "user_input": "",
+    },
+]
+
+
+async def run(key: str, user_input: str) -> None:
+    failures: list[str] = []
+
+    for scenario in SCENARIOS:
+        response = await graph(key, registry=global_registry).invoke(
+            user_input or scenario["user_input"],
+            new_context(),
+            {"user_id": "user-123"},
+            history=scenario["history"],
+        )
+
+        text = str(
+            response.get("response", "")
+            if isinstance(response, dict)
+            else getattr(response, "response", "")
+        )
+        saw_color = bool(re.search(r"\bred\b", text, re.IGNORECASE))
+
+        tag = "SAW" if saw_color else "DID NOT see"
+        print(
+            f"[graph-history-check] {scenario['name']}: model {tag} the image from history",
+            file=sys.stderr,
+        )
+        if not saw_color:
+            failures.append(scenario["name"])
+            print(
+                f"[graph-history-check] response was: {text[:300]}",
+                file=sys.stderr,
+            )
+
+        print(json.dumps(response, indent=2, default=str))
+        write_output(response)
+
+    if failures:
+        raise RuntimeError(
+            "graph() did not forward history to the root node for: "
+            + ", ".join(failures)
+            + ". Before the history feature lands this is the expected result."
+        )
diff --git a/examples/utils.py b/examples/utils.py
index f07f14fa..0f1c4333 100644
--- a/examples/utils.py
+++ b/examples/utils.py
@@ -2,10 +2,13 @@
 
 from __future__ import annotations
 
+import base64
 import dataclasses
 import json
 import random
 import string
+import struct
+import zlib
 from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
@@ -56,3 +59,30 @@ def write_output(data: Any) -> None:
         json.dumps(data, indent=2, default=_default_encoder), encoding="utf-8"
     )
     print(f"Output written to output/{filename}")
+
+
+def _png_chunk(kind: bytes, data: bytes) -> bytes:
+    return (
+        struct.pack(">I", len(data))
+        + kind
+        + data
+        + struct.pack(">I", zlib.crc32(kind + data) & 0xFFFFFFFF)
+    )
+
+
+def solid_color_png_base64(rgb: tuple[int, int, int], size: int = 64) -> str:
+    """Encodes a solid-colour PNG as base64 for multimodal examples.
+
+    Generating the image avoids committing a binary fixture, and the colour is
+    the only thing the model can report back — which makes it a usable signal
+    for whether the image actually reached the provider.
+    """
+    raw = b"".join(b"\x00" + bytes(rgb) * size for _ in range(size))
+    ihdr = struct.pack(">IIBBBBB", size, size, 8, 2, 0, 0, 0)
+    png = (
+        b"\x89PNG\r\n\x1a\n"
+        + _png_chunk(b"IHDR", ihdr)
+        + _png_chunk(b"IDAT", zlib.compress(raw))
+        + _png_chunk(b"IEND", b"")
+    )
+    return base64.b64encode(png).decode("ascii")
diff --git a/main.py b/main.py
index 66dd15a7..fc5c1e5b 100644
--- a/main.py
+++ b/main.py
@@ -9,6 +9,7 @@
     python main.py streaming                launch-darkly-documentation-summarizer "Summarise feature flags in 3 bullets"
     python main.py judge                    launch-darkly-documentation-summarizer "What is the LaunchDarkly AI SDK?"
     python main.py graph                    my-agent-graph "What is the LaunchDarkly AI SDK?"
+    python main.py graph-history            my-agent-graph ""
     python main.py openai-only              my-openai-flag "Tell me about feature flags"
     python main.py langchain                my-langchain-flag "Tell me about feature flags"
     python main.py claude-agents            launch-darkly-documentation-summarizer "What is the LaunchDarkly AI SDK?"
@@ -44,6 +45,7 @@
     "agent": "examples.agent",
     "streaming": "examples.streaming",
     "graph": "examples.graph_example",
+    "graph-history": "examples.graph_history",
     "conversation": "examples.conversation",
     "history": "examples.history",
     "judge": "examples.judge_example",
diff --git a/packages/claude-agents/src/launchdarkly_ai_claude_agents/handler.py b/packages/claude-agents/src/launchdarkly_ai_claude_agents/handler.py
index 182d31ec..32ee9fd6 100644
--- a/packages/claude-agents/src/launchdarkly_ai_claude_agents/handler.py
+++ b/packages/claude-agents/src/launchdarkly_ai_claude_agents/handler.py
@@ -35,7 +35,9 @@
     ProviderHandler,
     SpanMessage,
     SpanMessagePart,
+    compose_history,
     config,
+    content_to_text,
     create_handler,
     end_span_once,
     end_unfinished_spans,
@@ -320,17 +322,6 @@ def cancel_open_spans() -> None:
 # ---------------------------------------------------------------------------
 
 
-def _format_history(history: list[dict[str, Any]] | None) -> str | None:
-    if not history:
-        return None
-    lines = []
-    for msg in history:
-        role = msg.get("role", "user")
-        content = msg.get("content", "")
-        lines.append(f"{role}: {content}")
-    return "Conversation History:\n\n" + "\n".join(lines)
-
-
 def build_prompt(
     config: AiConfigRep,
     user_input: str | None,
@@ -362,15 +353,100 @@ def build_prompt(
             f"{config_history}\n\n{safe_input}" if config_history else safe_input
         )
 
-    history_text = _format_history(history)
-    if history_text:
-        system_prompt = (
-            f"{system_prompt}\n\n{history_text}" if system_prompt else history_text
-        )
-
     return safe_input, system_prompt
 
 
+def _parse_message_content(content: Any, variables: dict[str, Any]) -> Any:
+    return parse_template(content, variables) if isinstance(content, str) else content
+
+
+def _config_conversation_turns(
+    config: AiConfigRep, variables: dict[str, Any]
+) -> list[dict[str, Any]]:
+    return [
+        {
+            "role": message.get("role"),
+            "content": _parse_message_content(message.get("content", ""), variables),
+        }
+        for message in (config.get("messages") or [])
+        if message.get("role") != "system"
+    ]
+
+
+def _to_anthropic_user_content(content: Any) -> Any:
+    if isinstance(content, str):
+        return content
+
+    blocks: list[dict[str, Any]] = []
+    for block in content:
+        if block.get("type") == "text":
+            blocks.append({"type": "text", "text": block.get("text", "")})
+        elif block.get("type") == "image":
+            source = block.get("source", {})
+            if source.get("type") == "url":
+                mapped_source = {"type": "url", "url": source.get("url", "")}
+            else:
+                mapped_source = {
+                    "type": "base64",
+                    "media_type": source.get("media_type", ""),
+                    "data": source.get("data", ""),
+                }
+            blocks.append({"type": "image", "source": mapped_source})
+    return blocks
+
+
+async def _to_streamed_prompt(
+    turns: list[dict[str, Any]],
+) -> AsyncGenerator[dict[str, Any], None]:
+    # The envelope ``type`` has to agree with the message role. The CLI reading this stream
+    # accepts an "assistant" envelope as a replayed turn, but every other envelope type is
+    # required to carry role "user" — an assistant turn sent as ``type: "user"`` is rejected
+    # outright with "Expected message role 'user', got 'assistant'".
+    for turn in turns:
+        role = turn.get("role")
+        content = turn.get("content", "")
+        if role == "assistant":
+            yield {
+                "type": "assistant",
+                "message": {
+                    "role": "assistant",
+                    "content": [{"type": "text", "text": content_to_text(content)}],
+                },
+                "parent_tool_use_id": None,
+            }
+        else:
+            yield {
+                "type": "user",
+                "message": {
+                    "role": "user",
+                    "content": _to_anthropic_user_content(content),
+                },
+                "parent_tool_use_id": None,
+            }
+
+
+def build_query_prompt(
+    config: AiConfigRep,
+    user_input: str | None,
+    variables: dict[str, Any],
+    history: list[dict[str, Any]] | None,
+    fallback_prompt: str,
+) -> str | AsyncGenerator[dict[str, Any], None]:
+    if not history:
+        return fallback_prompt
+
+    turns = compose_history(
+        history=history,
+        user_input=user_input,
+        config_messages=(
+            []
+            if config.get("instructions")
+            else _config_conversation_turns(config, variables)
+        ),
+    )
+    return _to_streamed_prompt(turns)
+
+
 def _opening_of(prompt: str, system_prompt: str | None) -> Opening:
     return Opening(
         system_instructions=system_prompt,
@@ -451,6 +527,7 @@ async def _call_impl(
         open_root_span: Any = span
 
         prompt, system_prompt = build_prompt(config, user_input, vs, history)
+        query_prompt = build_query_prompt(config, user_input, vs, history, prompt)
         if config.get("outputFormat"):
             schema_instr = f"Respond with valid JSON matching this schema:\n{json.dumps(config['outputFormat'])}"
             system_prompt = (
@@ -510,7 +587,7 @@ async def _call_impl(
             # Held in a variable so the finally below can aclose() it. A bare `return` inside
             # `async for` abandons the generator, and asyncio's finalizer then raises RuntimeError
             # when the generator is suspended inside a real await in the SDK.
-            gen = query(prompt=prompt, options=options)
+            gen = query(prompt=query_prompt, options=options)
             try:
                 async for message in gen:
                     record_conversation_id(span, message)
@@ -647,6 +724,7 @@ async def _stream_gen(
     parent = parent_context_of(span)
 
     prompt, system_prompt = build_prompt(config, user_input, variables, history)
+    query_prompt = build_query_prompt(config, user_input, variables, history, prompt)
     opening = _opening_of(prompt, system_prompt)
 
     native_tool_map, user_config_tools, native_tool_names = partition_tools(
@@ -699,7 +777,7 @@ async def _stream_gen(
         )
 
         full_output = ""
-        gen = query(prompt=prompt, options=options)
+        gen = query(prompt=query_prompt, options=options)
         async for message in gen:
             record_conversation_id(span, message)
             record_native_tools(span, message, capture_content, catalog)
diff --git a/packages/claude-agents/src/launchdarkly_ai_claude_agents/native_graph.py b/packages/claude-agents/src/launchdarkly_ai_claude_agents/native_graph.py
index 0e0ed704..8fa8ce33 100644
--- a/packages/claude-agents/src/launchdarkly_ai_claude_agents/native_graph.py
+++ b/packages/claude-agents/src/launchdarkly_ai_claude_agents/native_graph.py
@@ -32,6 +32,7 @@
 from launchdarkly_ai_claude_agents.handler import (
     _build_hooks,
     build_prompt,
+    build_query_prompt,
     build_tool_mcp,
     partition_tools,
 )
@@ -103,6 +104,7 @@ async def _run_query(
     graph_key: str,
     run_id: str,
     child_subagent_tools: list[Any],
+    history: list[dict[str, Any]] | None = None,
 ) -> dict[str, Any]:
     import importlib
 
@@ -116,6 +118,9 @@ async def _run_query(
     wrapped = _wrap_native_tools(tool_handlers, ld_context, track_data)
 
     prompt, system_prompt = build_prompt(node.config, input_text, variables)
+    query_prompt = build_query_prompt(
+        node.config, input_text, variables, history, prompt
+    )
     native_tool_map, user_config_tools, native_tool_names = partition_tools(
         node.config.get("tools"), wrapped
     )
@@ -170,7 +175,7 @@ async def _run_query(
     # Bare `return` inside `async for` abandons the generator — Python's asyncio
     # finalizer later tries to aclose() it and may raise RuntimeError if the
     # generator is suspended inside a real await in the SDK (AIC-2950).
-    gen = query_fn(prompt=prompt, options=options)
+    gen = query_fn(prompt=query_prompt, options=options)
     try:
         async for message in gen:
             if isinstance(message, ResultMessage):
@@ -207,6 +212,7 @@ def to_claude_agents(
     async def invoke(
         input_text: str = "",
         variables: dict[str, Any] | None = None,
+        history: list[dict[str, Any]] | None = None,
     ) -> dict[str, Any]:
         import importlib
 
@@ -335,6 +341,7 @@ async def _subagent_execute(
                     def_obj.key,
                     run_id,
                     root_child_tools,
+                    history,
                 )
             except Exception as exc:
                 if span:
diff --git a/packages/claude-agents/tests/test_handler.py b/packages/claude-agents/tests/test_handler.py
index dc138c39..5651391e 100644
--- a/packages/claude-agents/tests/test_handler.py
+++ b/packages/claude-agents/tests/test_handler.py
@@ -318,10 +318,10 @@ def test_messages_mode_extracts_system_and_flattens_history_into_prompt(
         assert "context line" in prompt
         assert prompt.endswith("final question")
 
-    def test_history_appended_to_system_prompt(self) -> None:
+    def test_history_not_appended_to_system_prompt(self) -> None:
         history = [{"role": "user", "content": "earlier"}]
         _, system = build_prompt(BASE_CONFIG, "hi", {}, history=history)
-        assert "earlier" in (system or "")
+        assert system == "You are helpful."
         assert "You are helpful." in (system or "")
 
     def test_no_user_input_defaults_to_empty_string(self) -> None:
@@ -1399,20 +1399,24 @@ async def _query(**kwargs: Any) -> AsyncIterator[Any]:
 
 
 class TestHistoryAndVariables:
-    async def test_history_reaches_the_query_as_part_of_the_system_prompt(
+    async def test_history_reaches_query_as_structured_input(
         self, monkeypatch: pytest.MonkeyPatch
     ) -> None:
         captured: dict[str, Any] = {}
 
         async def _query(**kwargs: Any) -> AsyncIterator[Any]:
             captured["options"] = kwargs["options"]
+            captured["prompt"] = kwargs["prompt"]
             yield assistant_message()
             yield result_message()
 
         monkeypatch.setattr(handler_mod, "query", _query)
         history = [{"role": "user", "content": "earlier turn"}]
         await create_claude_agents_handler()(BASE_CONFIG, "q", history=history)
-        assert "earlier turn" in captured["options"].system_prompt
+        assert captured["options"].system_prompt == "You are helpful."
+        turns = [turn async for turn in captured["prompt"]]
+        assert turns[0]["message"]["content"] == "earlier turn"
+        assert turns[-1]["message"]["content"] == "q"
 
     async def test_ld_span_attributes_land_on_root_only(
         self, monkeypatch: pytest.MonkeyPatch
@@ -1713,12 +1717,12 @@ class TestHistory:
         {"role": "assistant", "content": "Feature flagging is a technique..."},
     ]
 
-    def test_history_format_is_correct(self) -> None:
-        config = _make_config(instructions="Be helpful.")
+    def test_history_not_stuffed_into_system_prompt(self) -> None:
+        config = _make_config(instructions="Be concise.")
         _, system = build_prompt(config, "hi", {}, self.SAMPLE_HISTORY)
         assert system is not None
-        assert "user: What is feature flagging?" in system
-        assert "assistant: Feature flagging is a technique..." in system
+        assert "Be concise." in system
+        assert "Conversation History:" not in system
 
     def test_empty_history_treated_like_no_history(self) -> None:
         config = _make_config(instructions="Be concise.")
@@ -1727,12 +1731,135 @@ def test_empty_history_treated_like_no_history(self) -> None:
         assert system_with_empty == system_without
         assert "Conversation History:" not in (system_with_empty or "")
 
-    def test_history_without_prior_system_prompt(self) -> None:
+    def test_history_without_instructions_keeps_system_none(self) -> None:
         config = _make_config()
         _, system = build_prompt(config, "hi", {}, self.SAMPLE_HISTORY)
-        assert system is not None
-        assert "Conversation History:" in system
-        assert "user: What is feature flagging?" in system
+        assert system is None or "Conversation History:" not in system
+
+
+IMAGE_BLOCK: dict[str, Any] = {
+    "type": "image",
+    "source": {"type": "base64", "media_type": "image/png", "data": "IMGDATA123"},
+}
+
+
+async def _capture_streamed_turns(
+    monkeypatch: pytest.MonkeyPatch,
+    user_input: str,
+    history: list[dict[str, Any]],
+) -> list[dict[str, Any]]:
+    """Runs the handler and returns the turns actually written to the SDK."""
+    captured: dict[str, Any] = {}
+
+    async def _query(**kwargs: Any) -> AsyncIterator[Any]:
+        prompt = kwargs["prompt"]
+        captured["turns"] = (
+            prompt if isinstance(prompt, str) else [turn async for turn in prompt]
+        )
+        yield result_message()
+
+    monkeypatch.setattr(handler_mod, "query", _query)
+    await create_claude_agents_handler()(BASE_CONFIG, user_input, history=history)
+    return captured["turns"]
+
+
+class TestMultimodalHistoryReachesTheProvider:
+    """The image has to arrive as a block, and the envelope has to be one the CLI accepts.
+
+    Asserting only that the run succeeded, or that the base64 payload appears somewhere in
+    the prompt, passes just as happily when the block has been flattened to a string on the
+    way out. These check the structure.
+    """
+
+    async def test_image_block_is_sent_as_a_block_not_stringified(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        turns = await _capture_streamed_turns(
+            monkeypatch,
+            "What colour is the square?",
+            [{"role": "user", "content": [IMAGE_BLOCK]}],
+        )
+
+        assert not isinstance(turns, str), "history collapsed to a single prompt string"
+        content = turns[0]["message"]["content"]
+        assert isinstance(content, list), f"content was stringified: {content!r}"
+        assert content == [
+            {
+                "type": "image",
+                "source": {
+                    "type": "base64",
+                    "media_type": "image/png",
+                    "data": "IMGDATA123",
+                },
+            }
+        ]
+
+    async def test_every_envelope_satisfies_the_cli_role_contract(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        # The CLI reading this stream accepts an "assistant" envelope as a replayed turn and
+        # rejects every other envelope whose role is not "user", with
+        # "Expected message role 'user', got 'assistant'".
+        turns = await _capture_streamed_turns(
+            monkeypatch,
+            "",
+            [
+                {"role": "user", "content": "I am going to share an image."},
+                {"role": "assistant", "content": "Sure — go ahead."},
+                {
+                    "role": "user",
+                    "content": [IMAGE_BLOCK, {"type": "text", "text": "What colour?"}],
+                },
+            ],
+        )
+
+        assert not isinstance(turns, str)
+        for turn in turns:
+            role = turn["message"]["role"]
+            assert turn["type"] == "assistant" or role == "user", (
+                f"envelope type {turn['type']!r} with role {role!r} is rejected by the CLI"
+            )
+            assert turn["type"] == role
+
+    async def test_assistant_turn_is_replayed_as_an_assistant_envelope(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        turns = await _capture_streamed_turns(
+            monkeypatch,
+            "",
+            [
+                {"role": "user", "content": "hi"},
+                {"role": "assistant", "content": "hello"},
+            ],
+        )
+
+        assert turns[1] == {
+            "type": "assistant",
+            "message": {
+                "role": "assistant",
+                "content": [{"type": "text", "text": "hello"}],
+            },
+            "parent_tool_use_id": None,
+        }
+
+    async def test_empty_user_input_appends_no_trailing_turn(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        # History that already ends with the user turn is sent as-is; an empty user_input
+        # must not be appended as a second, blank user turn.
+        turns = await _capture_streamed_turns(
+            monkeypatch,
+            "",
+            [
+                {"role": "user", "content": "hi"},
+                {"role": "assistant", "content": "hello"},
+                {"role": "user", "content": [IMAGE_BLOCK]},
+            ],
+        )
+
+        assert len(turns) == 3
+        assert turns[-1]["message"]["role"] == "user"
+        assert turns[-1]["message"]["content"] == [IMAGE_BLOCK]
 
 
 class TestBuiltinsSurviveAnEmptyToolList:
diff --git a/packages/claude-agents/tests/test_native_graph.py b/packages/claude-agents/tests/test_native_graph.py
index bf301ff7..da3195f2 100644
--- a/packages/claude-agents/tests/test_native_graph.py
+++ b/packages/claude-agents/tests/test_native_graph.py
@@ -276,6 +276,58 @@ async def test_runner_starts_at_root_and_returns_output(self) -> None:
 
         assert result["response"] == "final-output"
 
+    @pytest.mark.asyncio
+    async def test_multimodal_history_uses_native_root_prompt(self) -> None:
+        mock_sdk = _make_sdk_mock("done")
+        graph_def = _make_graph_def()
+        captured_prompts: list[Any] = []
+        result_msg = _make_result_msg("done")
+
+        async def _query(**kwargs: Any) -> AsyncIterator[Any]:
+            captured_prompts.append(kwargs.get("prompt"))
+            yield result_msg
+
+        mock_sdk.query = _query
+        history = [
+            {
+                "role": "user",
+                "content": [
+                    {
+                        "type": "image",
+                        "source": {
+                            "type": "base64",
+                            "media_type": "image/png",
+                            "data": "abc123",
+                        },
+                    }
+                ],
+            }
+        ]
+
+        with patch(
+            "importlib.import_module",
+            side_effect=lambda n: (
+                mock_sdk if n == "claude_agent_sdk" else __import__(n)
+            ),
+        ):
+            await to_claude_agents(_make_def_promise(graph_def)).invoke(
+                "describe", {}, history
+            )
+
+        assert captured_prompts
+        prompt = captured_prompts[0]
+        assert not isinstance(prompt, str)
+        chunks = [chunk async for chunk in prompt]
+        image = chunks[0]["message"]["content"][0]
+        assert image == {
+            "type": "image",
+            "source": {
+                "type": "base64",
+                "media_type": "image/png",
+                "data": "abc123",
+            },
+        }
+
     @pytest.mark.asyncio
     async def test_config_tools_converted_and_passed(self) -> None:
         mock_sdk = _make_sdk_mock("done")
diff --git a/packages/claude-messages/src/launchdarkly_ai_claude_messages/handler.py b/packages/claude-messages/src/launchdarkly_ai_claude_messages/handler.py
index 254a01e4..0de014a3 100644
--- a/packages/claude-messages/src/launchdarkly_ai_claude_messages/handler.py
+++ b/packages/claude-messages/src/launchdarkly_ai_claude_messages/handler.py
@@ -11,10 +11,13 @@
     ProviderHandler,
     SpanMessage,
     SpanMessagePart,
+    compose_history,
     config,
+    content_to_text,
     create_handler,
     end_span_once,
     end_unfinished_spans,
+    is_content_blocks,
     parse_template,
     set_input_content_attributes,
     set_output_content_attributes,
@@ -58,9 +61,65 @@ def _build_tools(config_tools: dict[str, Any]) -> list[dict[str, Any]]:
     ]
 
 
+def _anthropic_content(content: Any) -> str | list[dict[str, Any]]:
+    """Map LaunchDarkly-canonical content to Anthropic message content."""
+    if not is_content_blocks(content):
+        return content if isinstance(content, str) else ""
+
+    blocks: list[dict[str, Any]] = []
+    for block in content:
+        if block.get("type") == "text":
+            blocks.append({"type": "text", "text": block.get("text", "")})
+        elif block.get("type") == "image":
+            source = block.get("source", {})
+            if source.get("type") == "url":
+                mapped_source = {"type": "url", "url": source.get("url", "")}
+            else:
+                mapped_source = {
+                    "type": "base64",
+                    "media_type": source.get("media_type", ""),
+                    "data": source.get("data", ""),
+                }
+            blocks.append({"type": "image", "source": mapped_source})
+    return blocks
+
+
+def _template_content(content: Any, variables: dict[str, Any]) -> Any:
+    """Apply templates to text content without parsing structured blocks."""
+    return parse_template(content, variables) if isinstance(content, str) else content
+
+
+def _anthropic_blocks(content: Any) -> list[dict[str, Any]]:
+    """Normalize Anthropic message content to a list of content blocks."""
+    if isinstance(content, list):
+        return content
+    return [{"type": "text", "text": content}] if content else []
+
+
+def _merge_adjacent_same_role(
+    messages: list[dict[str, Any]],
+) -> list[dict[str, Any]]:
+    """Merge consecutive same-role turns into one multi-block message.
+
+    Anthropic's Messages API requires strictly alternating user/assistant roles.
+    Composed history can place an image-only user turn immediately before the
+    appended ``user_input`` question, which would otherwise send two consecutive
+    user turns and be rejected. Merging keeps both as a single user message.
+    """
+    merged: list[dict[str, Any]] = []
+    for message in messages:
+        if merged and merged[-1]["role"] == message["role"]:
+            merged[-1]["content"] = _anthropic_blocks(
+                merged[-1]["content"]
+            ) + _anthropic_blocks(message.get("content"))
+        else:
+            merged.append({"role": message["role"], "content": message.get("content")})
+    return merged
+
+
 def _build_messages(
     config: AiConfigRep,
-    user_input: str,
+    user_input: str | None,
     variables: dict[str, Any],
     *,
     include_output_format: bool = True,
@@ -68,33 +127,48 @@ def _build_messages(
 ) -> tuple[list[dict[str, Any]], str | None]:
     """Returns (messages, system_prompt)."""
     system: str | None = None
-    messages: list[dict[str, Any]] = []
+    config_messages: list[dict[str, Any]] = []
 
     if config.get("messages"):
         system_msgs = [m for m in config["messages"] if m.get("role") == "system"]
         conv_msgs = [m for m in config["messages"] if m.get("role") != "system"]
         if system_msgs:
             system = parse_template(
-                "\n".join(m["content"] for m in system_msgs), variables
+                "\n".join(content_to_text(m.get("content", "")) for m in system_msgs),
+                variables,
             )
         for msg in conv_msgs:
-            messages.append(
+            config_messages.append(
                 {
                     "role": msg["role"],
-                    "content": parse_template(msg["content"], variables),
+                    "content": _anthropic_content(
+                        _template_content(msg.get("content", ""), variables)
+                    ),
                 }
             )
     elif config.get("instructions"):
         system = parse_template(config["instructions"], variables)
 
     if history:
-        for msg in history:
-            role = msg.get("role", "user")
-            if role in ("user", "assistant"):
-                messages.append({"role": role, "content": msg.get("content", "")})
-
-    if not messages or messages[-1].get("role") != "user":
-        messages.append({"role": "user", "content": user_input or ""})
+        composed = compose_history(
+            history=history,
+            user_input=user_input,
+            config_messages=config_messages,
+        )
+        messages = _merge_adjacent_same_role(
+            [
+                {
+                    "role": msg["role"],
+                    "content": _anthropic_content(msg.get("content", "")),
+                }
+                for msg in composed
+                if msg.get("role") in ("user", "assistant")
+            ]
+        )
+    else:
+        messages = config_messages
+        if user_input or not messages:
+            messages.append({"role": "user", "content": user_input or ""})
 
     if include_output_format and config.get("outputFormat"):
         schema_instruction = f"Respond with valid JSON matching this schema:\n{json.dumps(config['outputFormat'])}"
diff --git a/packages/claude-messages/tests/test_handler.py b/packages/claude-messages/tests/test_handler.py
index 4c55c8e4..f07336fe 100644
--- a/packages/claude-messages/tests/test_handler.py
+++ b/packages/claude-messages/tests/test_handler.py
@@ -1733,6 +1733,64 @@ async def test_system_role_in_history_filtered_out(
         roles = [m["role"] for m in msgs]
         assert "system" not in roles
 
+    async def test_image_history_plus_user_input_merges_into_one_user_turn(
+        self, mock_anthropic: MagicMock
+    ) -> None:
+        from launchdarkly_ai_claude_messages import create_claude_messages_handler
+
+        history = [
+            {
+                "role": "user",
+                "content": [
+                    {
+                        "type": "image",
+                        "source": {
+                            "type": "base64",
+                            "media_type": "image/png",
+                            "data": "abc123",
+                        },
+                    }
+                ],
+            }
+        ]
+        h = create_claude_messages_handler()
+        await h(CONFIG, "What colour is this?", {}, {}, history)
+        msgs = mock_anthropic.messages.create.call_args.kwargs["messages"]
+        roles = [m["role"] for m in msgs]
+        assert not any(
+            roles[i] == "user" and roles[i + 1] == "user" for i in range(len(roles) - 1)
+        )
+        last = msgs[-1]
+        assert last["role"] == "user"
+        assert isinstance(last["content"], list)
+        assert any(b["type"] == "image" for b in last["content"])
+        assert any(
+            b["type"] == "text" and b["text"] == "What colour is this?"
+            for b in last["content"]
+        )
+
+    async def test_history_ending_in_user_text_plus_user_input_merges(
+        self, mock_anthropic: MagicMock
+    ) -> None:
+        from launchdarkly_ai_claude_messages import create_claude_messages_handler
+
+        h = create_claude_messages_handler()
+        await h(
+            CONFIG,
+            "follow-up",
+            {},
+            {},
+            [{"role": "user", "content": "prior question"}],
+        )
+        msgs = mock_anthropic.messages.create.call_args.kwargs["messages"]
+        merged_text = "".join(
+            b["text"]
+            for b in msgs[-1]["content"]
+            if isinstance(b, dict) and b.get("type") == "text"
+        )
+        assert "prior question" in merged_text
+        assert "follow-up" in merged_text
+
 
 class TestConvenienceWrapperForwardsCaptureContent:
     """`capture_content` must reach the handler, not fall through into `config()`.
diff --git a/packages/client/src/launchdarkly_ai_server/__init__.py b/packages/client/src/launchdarkly_ai_server/__init__.py
index e445229b..a2597e05 100644
--- a/packages/client/src/launchdarkly_ai_server/__init__.py
+++ b/packages/client/src/launchdarkly_ai_server/__init__.py
@@ -38,6 +38,14 @@
     init_evaluations,
 )
 from .graph import GraphInstance, graph, resolve_graph
+from .history import (
+    any_multimodal,
+    compose_history,
+    content_to_text,
+    has_multimodal_content,
+    image_block_to_url,
+    is_content_blocks,
+)
 from .judges import build_judge_tasks, run_judge, run_judges
 from .lifecycle import (
     extract_variation,
@@ -195,6 +203,13 @@
     "set_openllmetry_completion",
     "set_openllmetry_prompt",
     "to_ld_context",
+    # history
+    "compose_history",
+    "content_to_text",
+    "image_block_to_url",
+    "is_content_blocks",
+    "has_multimodal_content",
+    "any_multimodal",
     # validation
     "parse_ai_config",
     # registry
diff --git a/packages/client/src/launchdarkly_ai_server/graph.py b/packages/client/src/launchdarkly_ai_server/graph.py
index 95d1adee..e9088e71 100644
--- a/packages/client/src/launchdarkly_ai_server/graph.py
+++ b/packages/client/src/launchdarkly_ai_server/graph.py
@@ -207,6 +207,7 @@ async def run_node(
                 tool_handlers=tool_handlers,
                 variables=opts.get("variables"),
                 graph_key=key,
+                history=opts.get("history"),
             )
             response = (
                 result["response"]
@@ -359,6 +360,7 @@ def _fn(*a: Any, **kw: Any) -> str:
                 tool_handlers=merged_tool_handlers,
                 variables=opts.get("variables"),
                 graph_key=key,
+                history=opts.get("history"),
             )
             response = (
                 result["response"]
@@ -545,6 +547,7 @@ async def invoke(
         user_input: str | None,
         context: LDContext,
         variables: dict[str, Any] | None = None,
+        history: list[dict[str, Any]] | None = None,
     ) -> ProviderGraphResponse:
         from .judges import run_judges
         from .lifecycle import get_client
@@ -605,6 +608,11 @@ async def invoke(
                 opts: dict[str, Any] = {"variables": variables}
                 if previous_node:
                     opts["from"] = previous_node
+                # History seeds the entry point only. After the root hop, nodes
+                # stay oriented through the string threading built below, so
+                # history is not re-sent to downstream handlers.
+                elif history:
+                    opts["history"] = history
 
                 res = await graph_def.route(current, current_input, opts)
                 path.append(current.key)
diff --git a/packages/client/src/launchdarkly_ai_server/history.py b/packages/client/src/launchdarkly_ai_server/history.py
new file mode 100644
index 00000000..f5f46948
--- /dev/null
+++ b/packages/client/src/launchdarkly_ai_server/history.py
@@ -0,0 +1,97 @@
+"""Shared conversation-history composition and multimodal content helpers.
+
+Mirrors the TypeScript ``history`` module so every handler composes runtime
+``history`` the same way (TESTING.md §1.11) and maps LaunchDarkly-canonical
+content blocks to each provider's native shape (Appendix A.7).
+
+History messages are plain dicts: ``{"role": ..., "content": ...}`` where
+``content`` is either a string or a list of content-block dicts:
+
+    {"type": "text", "text": str}
+    {"type": "image", "source": {"type": "base64", "media_type": str, "data": str}}
+    {"type": "image", "source": {"type": "url", "url": str}}
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+MessageContent = str | list[dict[str, Any]]
+
+
+def compose_history(
+    *,
+    history: list[dict[str, Any]],
+    user_input: str | None = None,
+    config_messages: list[dict[str, Any]] | None = None,
+) -> list[dict[str, Any]]:
+    """Composes the ordered conversation turns for a handler with runtime history.
+
+    Order: ``[config conversation messages] -> [history] -> [user_input?]``.
+
+    - System-role history messages are dropped (system belongs on the provider
+      system prompt, derived separately by each handler).
+    - A non-empty ``user_input`` is always appended as a final user text turn,
+      even when history already ends with a user turn (image-only history + a
+      separate question).
+    - An empty/missing ``user_input`` appends nothing, so history that already
+      carries the full (possibly multimodal) user turn is sent as-is.
+
+    Returns a list of ``{"role": "user"|"assistant", "content": ...}`` dicts.
+    Callers only take this structured path when ``history`` is non-empty; with no
+    history they keep their single-string prompt behaviour, so an empty history
+    stays identical to passing none.
+    """
+    turns: list[dict[str, Any]] = list(config_messages or [])
+
+    for message in history:
+        role = message.get("role")
+        if role == "system":
+            continue
+        turns.append({"role": role, "content": message.get("content")})
+
+    if user_input:
+        turns.append({"role": "user", "content": user_input})
+
+    return turns
+
+
+def is_content_blocks(content: MessageContent) -> bool:
+    """True when content is the multimodal block-array shape."""
+    return isinstance(content, list)
+
+
+def has_multimodal_content(content: MessageContent) -> bool:
+    """True when a message carries any non-text (e.g. image) content block."""
+    if not isinstance(content, list):
+        return False
+    return any(block.get("type") != "text" for block in content)
+
+
+def any_multimodal(turns: list[dict[str, Any]]) -> bool:
+    """True when any turn in the list carries multimodal content."""
+    return any(has_multimodal_content(turn.get("content", "")) for turn in turns)
+
+
+def content_to_text(content: MessageContent) -> str:
+    """Flattens content to plain text: a string passes through; a block array
+    contributes only its text blocks."""
+    if isinstance(content, str):
+        return content
+    return "".join(
+        block.get("text", "") for block in content if block.get("type") == "text"
+    )
+
+
+def image_block_to_url(block: dict[str, Any]) -> str:
+    """Builds a ``data:<media_type>;base64,<data>`` URL for a base64 image block,
+    or returns the URL directly for a URL-sourced block.
+
+    This is the form OpenAI and LangChain expect (``image_url``); Anthropic keeps
+    ``media_type`` + ``data`` split, so its handlers read ``block["source"]``
+    directly instead.
+    """
+    source = block.get("source", {})
+    if source.get("type") == "url":
+        return str(source.get("url", ""))
+    return f"data:{source.get('media_type', '')};base64,{source.get('data', '')}"
diff --git a/packages/client/tests/test_graph.py b/packages/client/tests/test_graph.py
index 7264c063..69a752fd 100644
--- a/packages/client/tests/test_graph.py
+++ b/packages/client/tests/test_graph.py
@@ -423,3 +423,107 @@ async def failing_variation(key: str, ctx: dict, default: Any) -> Any:
 
         assert gd.enabled is False
         mock_logger.error.assert_called()
+
+    async def test_history_forwarded_to_root_handler_only(
+        self, mock_ld_client: MagicMock
+    ) -> None:
+        received: list[Any] = []
+
+        async def capturing_handler(
+            config: Any,
+            user_input: Any,
+            tool_handlers: Any,
+            variables: Any,
+            history: Any = None,
+        ) -> dict:
+            received.append(history)
+            return {"output": "ok", "usage": {"input_tokens": 1, "output_tokens": 1}}
+
+        handler = ProviderHandler(
+            fn=capturing_handler, provides_for=("TestProvider", "messages")
+        )  # type: ignore[arg-type]
+        history = [
+            {
+                "role": "user",
+                "content": [
+                    {
+                        "type": "image",
+                        "source": {
+                            "type": "base64",
+                            "media_type": "image/png",
+                            "data": "abc",
+                        },
+                    }
+                ],
+            }
+        ]
+        await graph("graph-key", handlers=[handler]).invoke(
+            "hi", CONTEXT, history=history
+        )
+        assert len(received) >= 2
+        assert received[0] == history
+        assert all(h is None for h in received[1:])
+
+    async def test_image_block_reaches_the_root_node_unstringified(
+        self, mock_ld_client: MagicMock
+    ) -> None:
+        # Guards the shape, not just presence: a root handler that received the image as a
+        # repr'd string instead of a block would still "contain" the data. History is
+        # root-only by design, so later nodes are expected to see none of it.
+        image_block = {
+            "type": "image",
+            "source": {
+                "type": "base64",
+                "media_type": "image/png",
+                "data": "IMGDATA123",
+            },
+        }
+        received: list[Any] = []
+
+        async def capturing_handler(
+            config: Any,
+            user_input: Any,
+            tool_handlers: Any,
+            variables: Any,
+            history: Any = None,
+        ) -> dict:
+            received.append(history)
+            return {"output": "ok", "usage": {"input_tokens": 1, "output_tokens": 1}}
+
+        handler = ProviderHandler(
+            fn=capturing_handler, provides_for=("TestProvider", "messages")
+        )  # type: ignore[arg-type]
+        await graph("graph-key", handlers=[handler]).invoke(
+            "what colour?",
+            CONTEXT,
+            history=[{"role": "user", "content": [image_block]}],
+        )
+
+        assert len(received) >= 2
+        root_content = received[0][0]["content"]
+        assert isinstance(root_content, list), (
+            f"content was stringified: {root_content!r}"
+        )
+        assert root_content == [image_block]
+        assert all(h is None for h in received[1:])
+
+    async def test_omitted_history_leaves_root_handler_history_none(
+        self, mock_ld_client: MagicMock
+    ) -> None:
+        received: list[Any] = []
+
+        async def capturing_handler(
+            config: Any,
+            user_input: Any,
+            tool_handlers: Any,
+            variables: Any,
+            history: Any = None,
+        ) -> dict:
+            received.append(history)
+            return {"output": "ok", "usage": {"input_tokens": 1, "output_tokens": 1}}
+
+        handler = ProviderHandler(
+            fn=capturing_handler, provides_for=("TestProvider", "messages")
+        )  # type: ignore[arg-type]
+        await graph("graph-key", handlers=[handler]).invoke("hi", CONTEXT)
+        assert received[0] is None
diff --git a/packages/langchain-agents/src/launchdarkly_ai_langchain_agents/handler.py b/packages/langchain-agents/src/launchdarkly_ai_langchain_agents/handler.py
index 682f0622..c2620c73 100644
--- a/packages/langchain-agents/src/launchdarkly_ai_langchain_agents/handler.py
+++ b/packages/langchain-agents/src/launchdarkly_ai_langchain_agents/handler.py
@@ -16,6 +16,7 @@
     ProviderHandler,
     SpanMessage,
     SpanMessagePart,
+    compose_history,
     config,
     create_handler,
     create_run_usage,
@@ -29,6 +30,7 @@
     set_output_content_attributes,
 )
 
+from .messages import to_lang_chain_messages
 from .spans import (
     build_span_callbacks,
     fail_span,
@@ -71,21 +73,9 @@ async def _handler(_name: str = name, **kwargs: Any) -> str:
     return result
 
 
-def _format_history(history: list[dict[str, Any]] | None) -> str | None:
-    if not history:
-        return None
-    lines = []
-    for msg in history:
-        role = msg.get("role", "user")
-        content = msg.get("content", "")
-        lines.append(f"{role}: {content}")
-    return "Conversation History:\n\n" + "\n".join(lines)
-
-
 def _extract_system_prompt(
     config: AiConfigRep,
     variables: dict[str, Any],
-    history: list[dict[str, Any]] | None = None,
 ) -> str | None:
     system_prompt: str | None = None
     if config.get("instructions"):
@@ -97,20 +87,43 @@ def _extract_system_prompt(
                 "\n".join(m["content"] for m in sys_msgs), variables
             )
 
-    history_text = _format_history(history)
-    if history_text:
-        system_prompt = (
-            f"{system_prompt}\n\n{history_text}" if system_prompt else history_text
-        )
-
     return system_prompt
 
 
+def _config_conversation_turns(
+    config: AiConfigRep, variables: dict[str, Any]
+) -> list[dict[str, Any]]:
+    return [
+        {
+            "role": message.get("role"),
+            "content": parse_template(message.get("content", ""), variables)
+            if isinstance(message.get("content", ""), str)
+            else message.get("content", ""),
+        }
+        for message in (config.get("messages") or [])
+        if message.get("role") != "system"
+    ]
+
+
 def _build_initial_messages(
     config: AiConfigRep,
     user_input: str,
     variables: dict[str, Any],
+    history: list[dict[str, Any]] | None = None,
 ) -> list[Any]:
+    if history:
+        return to_lang_chain_messages(
+            compose_history(
+                history=history,
+                user_input=user_input,
+                config_messages=(
+                    []
+                    if config.get("instructions")
+                    else _config_conversation_turns(config, variables)
+                ),
+            )
+        )
+
     import importlib
 
     msgs_mod = importlib.import_module("langchain_core.messages")
@@ -231,14 +244,14 @@ async def _call_impl(
         # truthily, and the test suite is built on mock spans.
         open_root_span: Any = span
 
-        system_prompt = _extract_system_prompt(config, vs, history)
+        system_prompt = _extract_system_prompt(config, vs)
         if config.get("outputFormat"):
             schema_instr = f"Respond with valid JSON matching this schema:\n{json.dumps(config['outputFormat'])}"
             system_prompt = (
                 f"{system_prompt}\n\n{schema_instr}" if system_prompt else schema_instr
             )
 
-        initial_messages = _build_initial_messages(config, user_input, vs)
+        initial_messages = _build_initial_messages(config, user_input, vs, history)
 
         span_callbacks = build_span_callbacks(
             config,
@@ -391,8 +404,8 @@ async def _stream_gen(
     span = start_root_span(config, variables)
     parent = parent_context_of(span)
 
-    system_prompt = _extract_system_prompt(config, variables, history)
-    initial_messages = _build_initial_messages(config, user_input, variables)
+    system_prompt = _extract_system_prompt(config, variables)
+    initial_messages = _build_initial_messages(config, user_input, variables, history)
 
     span_callbacks = build_span_callbacks(
         config, parent, capture_content, to_tool_definitions(config.get("tools") or {})
diff --git a/packages/langchain-agents/src/launchdarkly_ai_langchain_agents/messages.py b/packages/langchain-agents/src/launchdarkly_ai_langchain_agents/messages.py
new file mode 100644
index 00000000..f305df1e
--- /dev/null
+++ b/packages/langchain-agents/src/launchdarkly_ai_langchain_agents/messages.py
@@ -0,0 +1,54 @@
+"""Maps LaunchDarkly-canonical conversation turns onto LangChain messages.
+
+Images travel as ``image_url`` content parts with a data or remote URL — the
+standard multimodal shape every LangChain chat model accepts — rather than the
+LaunchDarkly-canonical ``{"type": "image", "source": ...}`` block, which no
+LangChain provider understands (TESTING.md Appendix A.7).
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from launchdarkly_ai_server import content_to_text, image_block_to_url
+
+
+def to_content_parts(content: str | list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """Maps one canonical message's content into LangChain user content parts."""
+    if isinstance(content, str):
+        return [{"type": "text", "text": content}]
+    parts: list[dict[str, Any]] = []
+    for block in content:
+        if block.get("type") == "text":
+            parts.append({"type": "text", "text": block.get("text", "")})
+        else:
+            parts.append(
+                {"type": "image_url", "image_url": {"url": image_block_to_url(block)}}
+            )
+    return parts
+
+
+def to_lang_chain_messages(turns: list[dict[str, Any]]) -> list[Any]:
+    """Turns composed canonical turns into LangChain messages.
+
+    A string user turn stays a string-content ``HumanMessage``, so text-only
+    callers see exactly the message they saw before history existed. Assistant
+    turns are flattened to text: an ``AIMessage`` carries the model's own prior
+    reply, which has no image to preserve.
+    """
+    import importlib
+
+    msgs_mod = importlib.import_module("langchain_core.messages")
+    HumanMessage = msgs_mod.HumanMessage
+    AIMessage = msgs_mod.AIMessage
+
+    messages: list[Any] = []
+    for turn in turns:
+        content = turn.get("content") or ""
+        if turn.get("role") == "assistant":
+            messages.append(AIMessage(content_to_text(content)))
+        elif isinstance(content, str):
+            messages.append(HumanMessage(content))
+        else:
+            messages.append(HumanMessage(to_content_parts(content)))
+    return messages
diff --git a/packages/langchain-agents/src/launchdarkly_ai_langchain_agents/native_graph.py b/packages/langchain-agents/src/launchdarkly_ai_langchain_agents/native_graph.py
index fd03223f..bf959881 100644
--- a/packages/langchain-agents/src/launchdarkly_ai_langchain_agents/native_graph.py
+++ b/packages/langchain-agents/src/launchdarkly_ai_langchain_agents/native_graph.py
@@ -15,12 +15,15 @@
     GraphDefinition,
     GraphNode,
     NativeTool,
+    compose_history,
     get_client,
     make_track_data,
     parse_template,
     to_ld_context,
 )
 
+from .messages import to_lang_chain_messages
+
 try:
     from opentelemetry import trace
     from opentelemetry.trace import StatusCode as SpanStatusCode
@@ -129,6 +132,7 @@ def to_lang_graph(
     async def invoke(
         input_text: str = "",
         variables: dict[str, Any] | None = None,
+        history: list[dict[str, Any]] | None = None,
     ) -> dict[str, Any]:
         import importlib
 
@@ -343,8 +347,19 @@ async def _pre_visit(node_key: str) -> None:
 
         compiled = builder.compile()
 
+        # History is a root-only concern: it seeds the initial message state the
+        # entry node reads. Downstream nodes are reached through handoffs and see
+        # the accumulated graph state, never the original `history` array.
+        initial_messages = (
+            to_lang_chain_messages(
+                compose_history(history=history, user_input=input_text)
+            )
+            if history
+            else [HumanMessage(input_text)]
+        )
+
         try:
-            result = await compiled.ainvoke({"messages": [HumanMessage(input_text)]})
+            result = await compiled.ainvoke({"messages": initial_messages})
             if span:
                 span.set_status(SpanStatusCode.OK)
         except Exception as exc:
diff --git a/packages/langchain-agents/tests/test_handler.py b/packages/langchain-agents/tests/test_handler.py
index 5165f026..a16b5eed 100644
--- a/packages/langchain-agents/tests/test_handler.py
+++ b/packages/langchain-agents/tests/test_handler.py
@@ -1516,33 +1516,84 @@ class TestHistory:
         {"role": "assistant", "content": "Feature flagging is a technique..."},
     ]
 
-    def test_history_appended_to_system_prompt(self) -> None:
+    def test_history_not_stuffed_into_system_prompt(self) -> None:
         config = _make_config(instructions="Be concise.")
-        system = _extract_system_prompt(config, {}, self.SAMPLE_HISTORY)
+        system = _extract_system_prompt(config, {})
         assert system is not None
-        assert "Conversation History:" in system
         assert "Be concise." in system
-
-    def test_history_format_is_correct(self) -> None:
-        config = _make_config(instructions="Be helpful.")
-        system = _extract_system_prompt(config, {}, self.SAMPLE_HISTORY)
-        assert system is not None
-        assert "user: What is feature flagging?" in system
-        assert "assistant: Feature flagging is a technique..." in system
+        assert "Conversation History:" not in system
 
     def test_empty_history_treated_like_no_history(self) -> None:
         config = _make_config(instructions="Be concise.")
-        system_with_empty = _extract_system_prompt(config, {}, [])
+        system_with_empty = _extract_system_prompt(config, {})
         system_without = _extract_system_prompt(config, {})
         assert system_with_empty == system_without
         assert "Conversation History:" not in (system_with_empty or "")
 
-    def test_history_without_prior_system_prompt(self) -> None:
+    def test_history_without_instructions_keeps_system_none(self) -> None:
         config = _make_config()
-        system = _extract_system_prompt(config, {}, self.SAMPLE_HISTORY)
-        assert system is not None
-        assert "Conversation History:" in system
-        assert "user: What is feature flagging?" in system
+        system = _extract_system_prompt(config, {})
+        assert system is None or "Conversation History:" not in system
+
+    @staticmethod
+    def _build(
+        config: dict[str, Any],
+        user_input: str | None,
+        history: list[dict[str, Any]] | None,
+    ) -> list[Any]:
+        lc_msgs = MagicMock()
+        lc_msgs.HumanMessage = MagicMock(
+            side_effect=lambda c: MagicMock(content=c, type="human")
+        )
+        lc_msgs.AIMessage = MagicMock(
+            side_effect=lambda c: MagicMock(content=c, type="ai")
+        )
+        with patch(
+            "importlib.import_module",
+            side_effect=lambda n: (
+                lc_msgs if n == "langchain_core.messages" else __import__(n)
+            ),
+        ):
+            return _build_initial_messages(config, user_input, {}, history)
+
+    def test_history_becomes_structured_messages_before_user_input(self) -> None:
+        msgs = self._build(
+            _make_config(instructions="Be concise."), "and now?", self.SAMPLE_HISTORY
+        )
+        assert [m.type for m in msgs] == ["human", "ai", "human"]
+        assert msgs[0].content == "What is feature flagging?"
+        assert msgs[-1].content == "and now?"
+
+    def test_image_history_maps_to_langchain_image_url_parts(self) -> None:
+        history = [
+            {
+                "role": "user",
+                "content": [
+                    {"type": "text", "text": "what is this?"},
+                    {
+                        "type": "image",
+                        "source": {
+                            "type": "base64",
+                            "media_type": "image/png",
+                            "data": "abc123",
+                        },
+                    },
+                ],
+            }
+        ]
+        msgs = self._build(_make_config(instructions="Be concise."), "", history)
+        assert {"type": "text", "text": "what is this?"} in msgs[0].content
+        assert {
+            "type": "image_url",
+            "image_url": {"url": "data:image/png;base64,abc123"},
+        } in msgs[0].content
+
+    def test_config_messages_precede_history(self) -> None:
+        config = _make_config(messages=[{"role": "user", "content": "config turn"}])
+        msgs = self._build(config, "q", self.SAMPLE_HISTORY)
+        assert msgs[0].content == "config turn"
+        assert msgs[1].content == "What is feature flagging?"
+        assert msgs[-1].content == "q"
 
 
 class TestAbandonOpenSpans:
diff --git a/packages/langchain-agents/tests/test_native_graph.py b/packages/langchain-agents/tests/test_native_graph.py
index c814618b..9191a537 100644
--- a/packages/langchain-agents/tests/test_native_graph.py
+++ b/packages/langchain-agents/tests/test_native_graph.py
@@ -5,6 +5,7 @@
 
 from __future__ import annotations
 
+import json
 import sys
 from contextlib import contextmanager
 from typing import Any
@@ -188,6 +189,38 @@ def compile(self) -> Any:
     }
 
 
+def _capture_compiled_state(mocks: dict[str, Any], ai_msg: Any) -> list[dict[str, Any]]:
+    """Swaps in a StateGraph whose compiled graph records the state it is invoked
+    with, so a test can assert on the root's initial messages."""
+    states: list[dict[str, Any]] = []
+
+    class _CapturingStateGraph:
+        def __init__(self, *a: Any, **kw: Any) -> None:
+            pass
+
+        def add_node(self, *a: Any, **kw: Any) -> None:
+            pass
+
+        def add_edge(self, *a: Any, **kw: Any) -> None:
+            pass
+
+        def add_conditional_edges(self, *a: Any, **kw: Any) -> None:
+            pass
+
+        def compile(self) -> Any:
+            compiled = MagicMock()
+
+            async def _ainvoke(state: dict[str, Any]) -> Any:
+                states.append(state)
+                return {"messages": [ai_msg]}
+
+            compiled.ainvoke = _ainvoke
+            return compiled
+
+    mocks["langgraph.graph"].StateGraph = _CapturingStateGraph
+    return states
+
+
 @contextmanager
 def _patch_imports(mocks: dict[str, Any]) -> Any:
     """Patch sys.modules so importlib.import_module picks up our mocks."""
@@ -730,6 +763,54 @@ async def _capture_invoke(msgs: list[Any]) -> Any:
         assert system_msgs, "No SystemMessage found"
         assert "expert" in system_msgs[0].content
 
+    @pytest.mark.asyncio
+    async def test_no_history_seeds_root_with_plain_human_message(self) -> None:
+        ai_msg = _make_ai_msg()
+        mocks = _make_langgraph_mocks(ai_msg)
+        states = _capture_compiled_state(mocks, ai_msg)
+        graph_def = _make_graph_def()
+
+        with _patch_imports(mocks):
+            await to_lang_graph(_make_def_promise(graph_def)).invoke("hi")
+
+        messages = states[0]["messages"]
+        assert len(messages) == 1
+        assert messages[0].content == "hi"
+
+    @pytest.mark.asyncio
+    async def test_history_seeds_root_with_native_image_content(self) -> None:
+        """A multimodal history reaches the root as LangChain image content."""
+        ai_msg = _make_ai_msg()
+        mocks = _make_langgraph_mocks(ai_msg)
+        states = _capture_compiled_state(mocks, ai_msg)
+        graph_def = _make_graph_def()
+
+        history = [
+            {
+                "role": "user",
+                "content": [
+                    {
+                        "type": "image",
+                        "source": {
+                            "type": "base64",
+                            "media_type": "image/png",
+                            "data": "abc123",
+                        },
+                    }
+                ],
+            }
+        ]
+
+        with _patch_imports(mocks):
+            await to_lang_graph(_make_def_promise(graph_def)).invoke(
+                "describe", {}, history
+            )
+
+        serialized = json.dumps([m.content for m in states[0]["messages"]])
+        assert "image_url" in serialized or '"type": "image"' in serialized
+        assert "abc123" in serialized
+        assert "describe" in serialized
+
     @pytest.mark.asyncio
     async def test_config_tools_creates_tool_node(self) -> None:
         """When config.tools is non-empty, a ToolNode must be created and wired."""
diff --git a/packages/langchain-messages/src/launchdarkly_ai_langchain_messages/handler.py b/packages/langchain-messages/src/launchdarkly_ai_langchain_messages/handler.py
index 8240da4a..54a466f4 100644
--- a/packages/langchain-messages/src/launchdarkly_ai_langchain_messages/handler.py
+++ b/packages/langchain-messages/src/launchdarkly_ai_langchain_messages/handler.py
@@ -13,11 +13,14 @@
     SpanMessage,
     SpanMessagePart,
     SpanUsage,
+    compose_history,
     config,
     create_handler,
     create_run_usage,
     end_span_once,
     end_unfinished_spans,
+    image_block_to_url,
+    is_content_blocks,
     lang_chain_content_text,
     lang_chain_finish_reasons,
     lang_chain_span_messages,
@@ -61,6 +64,23 @@ def _build_tools(config_tools: dict[str, Any]) -> list[dict[str, Any]]:
     ]
 
 
+def _to_langchain_content(content: Any) -> Any:
+    """Maps LD-canonical content blocks to LangChain multimodal content parts.
+    String content passes through so text-only callers keep plain strings."""
+    if not is_content_blocks(content):
+        return content if content is not None else ""
+
+    parts: list[dict[str, Any]] = []
+    for block in content:
+        if block.get("type") == "text":
+            parts.append({"type": "text", "text": block.get("text", "")})
+        elif block.get("type") == "image":
+            parts.append(
+                {"type": "image_url", "image_url": {"url": image_block_to_url(block)}}
+            )
+    return parts
+
+
 def _build_messages(
     config: AiConfigRep,
     user_input: str,
@@ -76,7 +96,7 @@ def _build_messages(
     AIMessage = msgs_mod.AIMessage
 
     messages: list[Any] = []
-    last_role: str | None = None
+    config_messages: list[dict[str, Any]] = []
 
     if config.get("messages"):
         system_msgs = [m for m in config["messages"] if m.get("role") == "system"]
@@ -90,28 +110,40 @@ def _build_messages(
                 )
             )
         for msg in conv_msgs:
-            content = parse_template(msg["content"], variables)
-            if msg["role"] == "user":
-                messages.append(HumanMessage(content))
-            elif msg["role"] == "assistant":
-                messages.append(AIMessage(content))
-            last_role = msg["role"]
+            content = msg.get("content", "")
+            if isinstance(content, str):
+                content = parse_template(content, variables)
+            if msg.get("role") in ("user", "assistant"):
+                config_messages.append({"role": msg["role"], "content": content})
     elif config.get("instructions"):
         messages.append(
             SystemMessage(parse_template(config["instructions"], variables))
         )
 
-    if history:
-        for msg in history:
-            role = msg.get("role", "user")
-            content = msg.get("content", "")
-            if role == "user":
-                messages.append(HumanMessage(content))
-            elif role == "assistant":
-                messages.append(AIMessage(content))
-            last_role = role
+    def append_turn(turn: dict[str, Any]) -> None:
+        content = _to_langchain_content(turn.get("content"))
+        if turn.get("role") == "user":
+            messages.append(HumanMessage(content=content))
+        else:
+            messages.append(AIMessage(content=content))
 
-    if last_role != "user":
+    if history:
+        for turn in compose_history(
+            history=history, user_input=user_input, config_messages=config_messages
+        ):
+            append_turn(turn)
+        return messages
+
+    for turn in config_messages:
+        append_turn(turn)
+
+    # Preserve the no-history behaviour: an empty input still produces a human
+    # message when the config carries no trailing user turn, so an empty history
+    # array stays identical to omitting history entirely.
+    last_non_system = next(
+        (m for m in reversed(messages) if getattr(m, "type", "") != "system"), None
+    )
+    if getattr(last_non_system, "type", "") != "human":
         messages.append(HumanMessage(user_input or ""))
     return messages
 
diff --git a/packages/langchain-messages/tests/test_handler.py b/packages/langchain-messages/tests/test_handler.py
index a260e013..dc0f7ccb 100644
--- a/packages/langchain-messages/tests/test_handler.py
+++ b/packages/langchain-messages/tests/test_handler.py
@@ -5,6 +5,7 @@
 
 from __future__ import annotations
 
+import json
 import sys
 from collections.abc import AsyncGenerator
 from typing import Any, ClassVar
@@ -1449,6 +1450,22 @@ class TestHistory:
         {"role": "user", "content": "What is feature flagging?"},
         {"role": "assistant", "content": "Feature flagging is a technique..."},
     ]
+    IMAGE_HISTORY: ClassVar[list[dict[str, Any]]] = [
+        {
+            "role": "user",
+            "content": [
+                {
+                    "type": "image",
+                    "source": {
+                        "type": "base64",
+                        "media_type": "image/png",
+                        "data": "abc123",
+                    },
+                },
+                {"type": "text", "text": "What is in this image?"},
+            ],
+        }
+    ]
 
     async def test_history_inserted_between_config_messages_and_user_input(
         self,
@@ -1506,6 +1523,28 @@ async def test_system_role_in_history_filtered_out(self) -> None:
         contents = [str(getattr(m, "content", "")) for m in call_args]
         assert "ignored" not in contents
 
+    async def test_multimodal_image_history_preserved_on_the_wire(self) -> None:
+        from launchdarkly_ai_langchain_messages import create_langchain_messages_handler
+
+        llm = _make_llm()
+        h = create_langchain_messages_handler(llm=llm)
+        await h(CONFIG, "", {}, {}, self.IMAGE_HISTORY)
+        call_args = llm.ainvoke.call_args[0][0]
+        serialized = json.dumps([getattr(m, "content", "") for m in call_args])
+        assert "image_url" in serialized
+        assert "abc123" in serialized
+
+    async def test_empty_user_input_with_history_ending_in_user(self) -> None:
+        from launchdarkly_ai_langchain_messages import create_langchain_messages_handler
+
+        llm = _make_llm()
+        h = create_langchain_messages_handler(llm=llm)
+        await h(CONFIG, "", {}, {}, [{"role": "user", "content": "Only turn"}])
+        call_args = llm.ainvoke.call_args[0][0]
+        humans = [m for m in call_args if getattr(m, "type", None) == "human"]
+        assert len(humans) == 1
+        assert humans[0].content == "Only turn"
+
 
 # ---------------------------------------------------------------------------
 # TELEMETRY-CONTRACT.md section 6: reported is not the same as reported zero
diff --git a/packages/openai-agents/src/launchdarkly_ai_openai_agents/handler.py b/packages/openai-agents/src/launchdarkly_ai_openai_agents/handler.py
index 45ac3117..faeee70c 100644
--- a/packages/openai-agents/src/launchdarkly_ai_openai_agents/handler.py
+++ b/packages/openai-agents/src/launchdarkly_ai_openai_agents/handler.py
@@ -29,11 +29,14 @@
     ProviderHandler,
     RunUsage,
     SpanUsage,
+    compose_history,
     config,
+    content_to_text,
     create_handler,
     create_run_usage,
     end_span_once,
     end_unfinished_spans,
+    image_block_to_url,
     parse_template,
     set_input_content_attributes,
     set_output_content_attributes,
@@ -107,15 +110,40 @@ async def _execute(_ctx: Any, args_str: str, _name: str = name) -> str:
     return result
 
 
-def _format_history(history: list[dict[str, Any]] | None) -> str | None:
-    if not history:
-        return None
-    lines = []
-    for msg in history:
-        role = msg.get("role", "user")
-        content = msg.get("content", "")
-        lines.append(f"{role}: {content}")
-    return "Conversation History:\n\n" + "\n".join(lines)
+def _parse_message_content(content: Any, variables: dict[str, Any]) -> Any:
+    """Apply templates to text content while preserving structured blocks."""
+    return parse_template(content, variables) if isinstance(content, str) else content
+
+
+def _to_openai_agent_items(turns: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """Map LaunchDarkly canonical turns to OpenAI Agents input items."""
+    items: list[dict[str, Any]] = []
+    for turn in turns:
+        role = turn["role"]
+        content = turn["content"]
+        if role == "assistant":
+            items.append({"role": "assistant", "content": content_to_text(content)})
+            continue
+
+        blocks = (
+            content
+            if isinstance(content, list)
+            else [{"type": "text", "text": content}]
+        )
+        parts: list[dict[str, Any]] = []
+        for block in blocks:
+            if block.get("type") == "image":
+                parts.append(
+                    {"type": "input_image", "image_url": image_block_to_url(block)}
+                )
+            elif block.get("type") == "text":
+                parts.append({"type": "input_text", "text": block.get("text", "")})
+        items.append({"role": "user", "content": parts})
+    return items
+
+
+def _prompt_to_text(prompt: str | list[dict[str, Any]]) -> str:
+    return prompt if isinstance(prompt, str) else json.dumps(prompt)
 
 
 def _build_agent_and_prompt(
@@ -124,7 +152,7 @@ def _build_agent_and_prompt(
     tool_handlers: dict[str, Any],
     variables: dict[str, Any],
     history: list[dict[str, Any]] | None = None,
-) -> tuple[Any, str, str | None]:
+) -> tuple[Any, str | list[dict[str, Any]], str | None]:
     import importlib
 
     agents_mod = importlib.import_module("agents")
@@ -132,27 +160,46 @@ def _build_agent_and_prompt(
 
     safe_input = user_input or ""
     instructions: str | None = None
-    prompt = safe_input
+    prompt: str | list[dict[str, Any]] = safe_input
+
+    config_messages = config.get("messages") or []
+    parsed_messages = [
+        {
+            **message,
+            "content": _parse_message_content(message.get("content", ""), variables),
+        }
+        for message in config_messages
+    ]
 
     if config.get("instructions"):
         instructions = parse_template(config["instructions"], variables)
-    elif config.get("messages"):
-        system_msgs = [m for m in config["messages"] if m.get("role") == "system"]
-        conv_msgs = [m for m in config["messages"] if m.get("role") != "system"]
+    elif parsed_messages:
+        system_msgs = [m for m in parsed_messages if m.get("role") == "system"]
+        conv_msgs = [m for m in parsed_messages if m.get("role") != "system"]
         if system_msgs:
-            instructions = parse_template(
-                "\n".join(m["content"] for m in system_msgs), variables
-            )
-        conv_history = "\n".join(
-            parse_template(m["content"], variables) for m in conv_msgs
-        )
+            instructions = "\n".join(content_to_text(m["content"]) for m in system_msgs)
+        conv_history = "\n".join(content_to_text(m["content"]) for m in conv_msgs)
         prompt = f"{conv_history}\n\n{safe_input}" if conv_history else safe_input
 
-    history_text = _format_history(history)
-    if history_text:
-        instructions = (
-            f"{instructions}\n\n{history_text}" if instructions else history_text
+    if history:
+        # When config.instructions is set, config.messages conversation turns are
+        # ignored (see the no-history branches above), so history composition must
+        # not resurrect them — mirror that priority here.
+        config_history_messages = (
+            []
+            if config.get("instructions")
+            else [
+                message
+                for message in parsed_messages
+                if message.get("role") != "system"
+            ]
+        )
+        turns = compose_history(
+            history=history,
+            user_input=user_input,
+            config_messages=config_history_messages,
         )
+        prompt = _to_openai_agent_items(turns)
 
     tools = _build_agent_tools(config.get("tools") or {}, tool_handlers)
 
@@ -449,7 +496,7 @@ async def _call_impl(
                 span,
                 capture_content,
                 system_instructions=instructions,
-                messages=[text_message("user", prompt)],
+                messages=to_request_span_messages(prompt),
             )
             result = await Runner.run(agent, prompt, hooks=hooks)
             final_output = result.final_output
@@ -579,7 +626,7 @@ async def _stream_gen(
             span,
             capture_content,
             system_instructions=instructions,
-            messages=[text_message("user", prompt)],
+            messages=to_request_span_messages(prompt),
         )
         streamed = Runner.run_streamed(agent, prompt, hooks=hooks)
         full_output = ""
diff --git a/packages/openai-agents/src/launchdarkly_ai_openai_agents/native_graph.py b/packages/openai-agents/src/launchdarkly_ai_openai_agents/native_graph.py
index f66b1c1b..88aef9cb 100644
--- a/packages/openai-agents/src/launchdarkly_ai_openai_agents/native_graph.py
+++ b/packages/openai-agents/src/launchdarkly_ai_openai_agents/native_graph.py
@@ -15,12 +15,15 @@
     GraphDefinition,
     GraphNode,
     NativeTool,
+    compose_history,
     get_client,
     make_track_data,
     parse_template,
     to_ld_context,
 )
 
+from .handler import _parse_message_content, _to_openai_agent_items
+
 try:
     from opentelemetry import trace
     from opentelemetry.trace import StatusCode as SpanStatusCode
@@ -99,6 +102,7 @@ def to_openai_agents(
     async def invoke(
         input_text: str = "",
         variables: dict[str, Any] | None = None,
+        history: list[dict[str, Any]] | None = None,
     ) -> dict[str, Any]:
         import importlib
 
@@ -216,8 +220,34 @@ async def on_agent_start(self, context: Any, agent: Any) -> None:
 
         hooks = _LDHooks()
 
+        root_prompt: str | list[dict[str, Any]] = input_text
+        if history:
+            # config.instructions takes priority over config.messages, so skip
+            # config conversation turns when instructions are set (parity with the
+            # single-node handler and TESTING.md §1.11 composition order).
+            config_messages = (
+                []
+                if root.config.get("instructions")
+                else [
+                    {
+                        **message,
+                        "content": _parse_message_content(
+                            message.get("content", ""), vs
+                        ),
+                    }
+                    for message in (root.config.get("messages") or [])
+                    if message.get("role") != "system"
+                ]
+            )
+            turns = compose_history(
+                history=history,
+                user_input=input_text,
+                config_messages=config_messages,
+            )
+            root_prompt = _to_openai_agent_items(turns)
+
         try:
-            result = await Runner.run(root_agent, input_text, hooks=hooks)
+            result = await Runner.run(root_agent, root_prompt, hooks=hooks)
             if span:
                 span.set_status(SpanStatusCode.OK)
         except Exception as exc:
diff --git a/packages/openai-agents/tests/test_handler.py b/packages/openai-agents/tests/test_handler.py
index 73c9e78e..4268ee01 100644
--- a/packages/openai-agents/tests/test_handler.py
+++ b/packages/openai-agents/tests/test_handler.py
@@ -1484,40 +1484,65 @@ class TestHistory:
         {"role": "user", "content": "What is feature flagging?"},
         {"role": "assistant", "content": "Feature flagging is a technique..."},
     ]
+    IMAGE_HISTORY: ClassVar[list[dict[str, Any]]] = [
+        {
+            "role": "user",
+            "content": [
+                {
+                    "type": "image",
+                    "source": {
+                        "type": "base64",
+                        "media_type": "image/png",
+                        "data": "abc123",
+                    },
+                }
+            ],
+        }
+    ]
 
-    def test_history_appended_to_instructions(self) -> None:
+    def test_history_is_structured_input_not_system_prompt_text(self) -> None:
         config = _make_config(instructions="Be concise.")
-        _, _, instructions = _build_agent_and_prompt(
+        _, prompt, instructions = _build_agent_and_prompt(
             config, "hi", {}, {}, self.SAMPLE_HISTORY
         )
         assert instructions is not None
-        assert "Conversation History:" in instructions
         assert "Be concise." in instructions
+        assert "Conversation History:" not in instructions
+        assert isinstance(prompt, list)
+        assert "What is feature flagging?" in str(prompt)
+        assert "hi" in str(prompt)
 
-    def test_history_format_is_correct(self) -> None:
+    def test_history_turns_appear_before_user_input(self) -> None:
         config = _make_config(instructions="Be helpful.")
-        _, _, instructions = _build_agent_and_prompt(
-            config, "hi", {}, {}, self.SAMPLE_HISTORY
+        _, prompt, _ = _build_agent_and_prompt(
+            config, "follow up", {}, {}, self.SAMPLE_HISTORY
+        )
+        assert isinstance(prompt, list)
+        serialized = str(prompt)
+        assert serialized.index("What is feature flagging?") < serialized.rindex(
+            "follow up"
         )
-        assert instructions is not None
-        assert "user: What is feature flagging?" in instructions
-        assert "assistant: Feature flagging is a technique..." in instructions
 
     def test_empty_history_treated_like_no_history(self) -> None:
         config = _make_config(instructions="Be concise.")
-        _, _, instr_with_empty = _build_agent_and_prompt(config, "hi", {}, {}, [])
-        _, _, instr_without = _build_agent_and_prompt(config, "hi", {}, {})
+        _, prompt_with_empty, instr_with_empty = _build_agent_and_prompt(
+            config, "hi", {}, {}, []
+        )
+        _, prompt_without, instr_without = _build_agent_and_prompt(config, "hi", {}, {})
         assert instr_with_empty == instr_without
         assert "Conversation History:" not in (instr_with_empty or "")
+        assert prompt_with_empty == prompt_without
 
-    def test_history_without_prior_instructions(self) -> None:
-        config = _make_config()
-        _, _, instructions = _build_agent_and_prompt(
-            config, "hi", {}, {}, self.SAMPLE_HISTORY
+    def test_multimodal_image_history_maps_to_input_image(self) -> None:
+        config = _make_config(instructions="Be helpful.")
+        _, prompt, instructions = _build_agent_and_prompt(
+            config, "describe", {}, {}, self.IMAGE_HISTORY
         )
         assert instructions is not None
-        assert "Conversation History:" in instructions
-        assert "user: What is feature flagging?" in instructions
+        assert "Conversation History:" not in instructions
+        assert isinstance(prompt, list)
+        assert "input_image" in str(prompt)
+        assert "data:image/png;base64,abc123" in str(prompt)
 
 
 # ---------------------------------------------------------------------------
diff --git a/packages/openai-agents/tests/test_native_graph.py b/packages/openai-agents/tests/test_native_graph.py
index a30dbb38..d596a425 100644
--- a/packages/openai-agents/tests/test_native_graph.py
+++ b/packages/openai-agents/tests/test_native_graph.py
@@ -186,6 +186,41 @@ async def test_root_node_is_entry_point(self) -> None:
         call_args = agents_mock.Runner.run.call_args
         assert "input-text" in call_args[0] or "input-text" == call_args[0][1]
 
+    @pytest.mark.asyncio
+    async def test_multimodal_history_is_structured_root_input(self) -> None:
+        run_result = _make_run_result("out")
+        agents_mock = _make_agents_mock(run_result)
+        graph_def = _make_graph_def()
+        history = [
+            {
+                "role": "user",
+                "content": [
+                    {
+                        "type": "image",
+                        "source": {
+                            "type": "base64",
+                            "media_type": "image/png",
+                            "data": "abc123",
+                        },
+                    }
+                ],
+            }
+        ]
+
+        with patch(
+            "importlib.import_module",
+            side_effect=lambda n: agents_mock if n == "agents" else __import__(n),
+        ):
+            await to_openai_agents(_make_def_promise(graph_def)).invoke(
+                "describe", None, history
+            )
+
+        root_input = agents_mock.Runner.run.call_args.args[1]
+        assert isinstance(root_input, list)
+        serialized = str(root_input)
+        assert "input_image" in serialized
+        assert "data:image/png;base64,abc123" in serialized
+
     @pytest.mark.asyncio
     async def test_terminal_nodes_no_handoff_tools(self) -> None:
         run_result = _make_run_result("out")
diff --git a/packages/openai-messages/src/launchdarkly_ai_openai_messages/handler.py b/packages/openai-messages/src/launchdarkly_ai_openai_messages/handler.py
index 0d651eb5..a404fdcb 100644
--- a/packages/openai-messages/src/launchdarkly_ai_openai_messages/handler.py
+++ b/packages/openai-messages/src/launchdarkly_ai_openai_messages/handler.py
@@ -12,11 +12,15 @@
     RunUsage,
     SpanMessage,
     SpanMessagePart,
+    compose_history,
     config,
+    content_to_text,
     create_handler,
     create_run_usage,
     end_span_once,
     end_unfinished_spans,
+    image_block_to_url,
+    is_content_blocks,
     parse_template,
     set_input_content_attributes,
     set_output_content_attributes,
@@ -65,30 +69,72 @@ def _build_input_messages(
     variables: dict[str, Any],
     history: list[dict[str, Any]] | None = None,
 ) -> list[dict[str, Any]]:
+    system_messages: list[dict[str, Any]] = []
+    config_messages: list[dict[str, Any]] = []
+
     if config.get("messages"):
-        msgs = [
-            {"role": m["role"], "content": parse_template(m["content"], variables)}
-            for m in config["messages"]
-        ]
-        if history:
-            for msg in history:
-                role = msg.get("role", "user")
-                if role in ("user", "assistant"):
-                    msgs.append({"role": role, "content": msg.get("content", "")})
-        if user_input and (not msgs or msgs[-1].get("role") != "user"):
-            msgs.append({"role": "user", "content": user_input})
-        return msgs
-    instructions = parse_template(config.get("instructions") or "", variables)
-    result: list[dict[str, Any]] = []
-    if instructions:
-        result.append({"role": "system", "content": instructions})
+        for message in config["messages"]:
+            content = message.get("content", "")
+            mapped = {
+                "role": message["role"],
+                "content": parse_template(content, variables)
+                if isinstance(content, str)
+                else content,
+            }
+            if message["role"] == "system":
+                system_messages.append(mapped)
+            else:
+                config_messages.append(mapped)
+    else:
+        instructions = parse_template(config.get("instructions") or "", variables)
+        if instructions:
+            system_messages.append({"role": "system", "content": instructions})
+
     if history:
-        for msg in history:
-            role = msg.get("role", "user")
-            if role in ("user", "assistant"):
-                result.append({"role": role, "content": msg.get("content", "")})
-    result.append({"role": "user", "content": user_input or ""})
-    return result
+        turns = compose_history(
+            history=history,
+            user_input=user_input,
+            config_messages=config_messages,
+        )
+    else:
+        turns = list(config_messages)
+        if config.get("messages"):
+            if user_input and (not turns or turns[-1].get("role") != "user"):
+                turns.append({"role": "user", "content": user_input})
+        else:
+            turns.append({"role": "user", "content": user_input or ""})
+
+    return system_messages + [
+        {"role": turn["role"], "content": _openai_content(turn)}
+        for turn in turns
+        if turn.get("role") in ("user", "assistant")
+    ]
+
+
+def _openai_content(message: dict[str, Any]) -> Any:
+    """Map canonical content blocks to Responses API input content."""
+    raw_content = message.get("content")
+    content: str | list[dict[str, Any]] = (
+        raw_content if isinstance(raw_content, (str, list)) else ""
+    )
+    if not is_content_blocks(content):
+        return content
+    assert isinstance(content, list)
+
+    # ``input_text``/``input_image`` are the input-side part types, and the Responses API
+    # only accepts them on a user turn. A replayed assistant turn flattens to its text.
+    if message.get("role") != "user":
+        return content_to_text(content)
+
+    parts: list[dict[str, Any]] = []
+    for block in content:
+        if block.get("type") == "text":
+            parts.append({"type": "input_text", "text": block.get("text", "")})
+        elif block.get("type") == "image":
+            parts.append(
+                {"type": "input_image", "image_url": image_block_to_url(block)}
+            )
+    return parts
 
 
 def _json_schema_format(schema: dict[str, Any]) -> dict[str, Any]:
diff --git a/packages/openai-messages/tests/test_handler.py b/packages/openai-messages/tests/test_handler.py
index 11651ddb..4668976f 100644
--- a/packages/openai-messages/tests/test_handler.py
+++ b/packages/openai-messages/tests/test_handler.py
@@ -1745,6 +1745,61 @@ async def test_history_with_instructions_path(self, mock_openai: MagicMock) -> N
         assert non_system[1]["content"] == "Feature flagging is a technique..."
         assert non_system[-1]["content"] == "my question"
 
+    async def test_user_image_block_becomes_an_input_image_part(
+        self, mock_openai: MagicMock
+    ) -> None:
+        from launchdarkly_ai_openai_messages import create_openai_messages_handler
+
+        image_block = {
+            "type": "image",
+            "source": {
+                "type": "base64",
+                "media_type": "image/png",
+                "data": "IMGDATA123",
+            },
+        }
+        h = create_openai_messages_handler()
+        await h(
+            CONFIG, "what colour?", {}, {}, [{"role": "user", "content": [image_block]}]
+        )
+        msgs = mock_openai.responses.create.call_args.kwargs["input"]
+        user_msg = next(m for m in msgs if m.get("role") == "user")
+
+        assert isinstance(user_msg["content"], list), (
+            f"content was stringified: {user_msg['content']!r}"
+        )
+        assert user_msg["content"] == [
+            {
+                "type": "input_image",
+                "image_url": "data:image/png;base64,IMGDATA123",
+            }
+        ]
+
+    async def test_assistant_block_content_flattens_to_text(
+        self, mock_openai: MagicMock
+    ) -> None:
+        # ``input_text`` is an input-side part type; the Responses API only accepts it on a
+        # user turn, so a replayed assistant turn has to arrive as a plain string.
+        from launchdarkly_ai_openai_messages import create_openai_messages_handler
+
+        h = create_openai_messages_handler()
+        await h(
+            CONFIG,
+            "q",
+            {},
+            {},
+            [
+                {"role": "user", "content": "hi"},
+                {
+                    "role": "assistant",
+                    "content": [{"type": "text", "text": "prior answer"}],
+                },
+            ],
+        )
+        msgs = mock_openai.responses.create.call_args.kwargs["input"]
+        assistant_msg = next(m for m in msgs if m.get("role") == "assistant")
+        assert assistant_msg["content"] == "prior answer"
+
     async def test_empty_history_treated_like_no_history(
         self, mock_openai: MagicMock
     ) -> None:
@@ -1781,6 +1836,16 @@ async def test_system_role_in_history_filtered_out(
         ]
         assert "system" not in history_roles
 
+    async def test_empty_user_input_no_history_still_sends_user_turn(
+        self, mock_openai: MagicMock
+    ) -> None:
+        from launchdarkly_ai_openai_messages import create_openai_messages_handler
+
+        h = create_openai_messages_handler()
+        await h(CONFIG, "", {}, {})
+        msgs = mock_openai.responses.create.call_args.kwargs["input"]
+        assert any(m.get("role") == "user" for m in msgs)
+
 
 class TestConvenienceWrapperForwardsCaptureContent:
     """`capture_content` must reach the handler, not fall through into `config()`.