system (from instructions)
system (from developer message 1)
system (from developer message 2)
...
user
user

---

from typing import Literal, Optional, Union, Any
from litellm.integrations.custom_logger import CustomLogger
from litellm.proxy._types import UserAPIKeyAuth
from litellm.caching.caching import DualCache


class MergeSystemMessagesHook(CustomLogger):
    def _extract_text(self, content: Any) -> str:
        if isinstance(content, str):
            return content
        if isinstance(content, list):
            parts = []
            for block in content:
                if isinstance(block, dict):
                    t = block.get("type", "")
                    if t in ("text", "input_text"):
                        parts.append(block.get("text", ""))
            return "\n\n".join(p for p in parts if p)
        return ""

    async def async_pre_call_hook(
        self,
        user_api_key_dict: UserAPIKeyAuth,
        cache: DualCache,
        data: dict,
        call_type: Literal[
            "completion", "text_completion", "embeddings", "image_generation",
            "moderation", "audio_transcription", "pass_through_endpoint",
            "rerank", "mcp_call", "anthropic_messages",
        ],
    ) -> Optional[Union[Exception, str, dict]]:
        if call_type == "aresponses":
            return self._merge_responses_api(data)
        if call_type == "completion":
            return self._merge_completion_messages(data)
        return data

    def _merge_responses_api(self, data: dict) -> dict:
        input_items = data.get("input") or []
        if not isinstance(input_items, list):
            return data
        developer_parts, new_input = [], []
        for item in input_items:
            if isinstance(item, dict) and item.get("role") == "developer":
                text = self._extract_text(item.get("content", ""))
                if text:
                    developer_parts.append(text)
            else:
                new_input.append(item)
        if not developer_parts:
            return data
        existing = data.get("instructions", "") or ""
        all_parts = ([existing] if existing else []) + developer_parts
        data["instructions"] = "\n\n".join(all_parts)
        data["input"] = new_input
        return data

    def _merge_completion_messages(self, data: dict) -> dict:
        messages = data.get("messages")
        if not messages:
            return data
        system_parts, rest = [], []
        for msg in messages:
            if msg.get("role") == "system":
                text = self._extract_text(msg.get("content", ""))
                if text:
                    system_parts.append(text)
            else:
                rest.append(msg)
        if len(system_parts) <= 1:
            return data
        data["messages"] = [{"role": "system", "content": "\n\n".join(system_parts)}] + rest
        return data


merge_system_messages_hook = MergeSystemMessagesHook()

---

litellm_settings:
  callbacks:
    - merge_system_messages.merge_system_messages_hook

---