feat(engine): add generic extra_headers support for LLM providers

Add extra_headers field to ProviderConfig, ResolvedLLMConfig, and
LLMCallInput. Values use {ENV_VAR} syntax for runtime resolution.
Headers are merged into _call_openai() (default_headers), _call_anthropic()
(headers dict), and EmbeddingExecutor (default_headers).

This enables config-driven proxy routing where the YAML generator
injects X-Org-Id and X-User-Id headers without engine-specific code.

TASK-207

Author: ibacalu

src/workflows_mcp/engine/executors_llm.py

@@ -10,6 +10,7 @@
 import asyncio
 import json
 import logging
+import os
 from enum import Enum
 from typing import Any, ClassVar, cast
 
@@ -50,6 +51,21 @@ class LLMProvider(str, Enum):
     OLLAMA = "ollama"
 
 
+def _resolve_header_env_vars(headers: dict[str, str]) -> dict[str, str]:
+    """Resolve {ENV_VAR} placeholders in header values.
+
+    Values wrapped in curly braces (e.g., "{ORG_ID}") are resolved from
+    environment variables at runtime. Other values are passed through as-is.
+    """
+    resolved = {}
+    for k, v in headers.items():
+        if v.startswith("{") and v.endswith("}"):
+            resolved[k] = os.environ.get(v[1:-1], "")
+        else:
+            resolved[k] = v
+    return resolved
+
+
 # ===========================================================================
 # LLMCall Executor
 # ===========================================================================
@@ -110,6 +126,10 @@ class LLMCallInput(BlockInput):
         default=None,
         description="Custom API endpoint URL (optional, for custom deployments)",
     )
+    extra_headers: dict[str, str] = Field(
+        default_factory=dict,
+        description="Custom HTTP headers for provider requests (resolved from config)",
+    )
     response_schema: dict[str, Any] | str | None = Field(
         default=None,
         description=(
@@ -604,6 +624,7 @@ async def _resolve_profile_to_inputs(
             temperature=resolved_config.temperature,
             max_tokens=resolved_config.max_tokens,
             validation_prompt_template=inputs.validation_prompt_template,
+            extra_headers=resolved_config.extra_headers,
         )
 
     def _resolve_profile_with_fallback(
@@ -915,6 +936,10 @@ async def _call_openai(
                 base_url = base_url.rsplit("/chat/completions", 1)[0]
             client_kwargs["base_url"] = base_url
 
+        # Merge extra_headers (e.g., X-Org-Id, X-User-Id for proxy routing)
+        if inputs.extra_headers:
+            client_kwargs["default_headers"] = _resolve_header_env_vars(inputs.extra_headers)
+
         # Prepare completion parameters (required parameters only)
         completion_kwargs: dict[str, Any] = {
             "model": inputs.model or "",
@@ -1046,6 +1071,10 @@ async def _call_anthropic(
         if inputs.api_key:
             headers["x-api-key"] = inputs.api_key
 
+        # Merge extra_headers (e.g., X-Org-Id, X-User-Id for proxy routing)
+        if inputs.extra_headers:
+            headers.update(_resolve_header_env_vars(inputs.extra_headers))
+
         async with httpx.AsyncClient(timeout=timeout) as client:
             response = await client.post(url, json=body, headers=headers)
             response.raise_for_status()
@@ -1428,16 +1457,25 @@ async def execute(  # type: ignore[override]
         if model is None:
             model = "text-embedding-3-small"
 
+        # Resolve extra_headers from profile config
+        default_headers: dict[str, str] | None = None
+        if resolved_config and resolved_config.extra_headers:
+            default_headers = _resolve_header_env_vars(resolved_config.extra_headers)
+
         # Resolve timeout
         timeout = resolve_interpolatable_numeric(inputs.timeout, int, "timeout", ge=1, le=300)
 
         try:
             # Use OpenAI SDK which works with any OpenAI-compatible server
-            client = AsyncOpenAI(
-                api_key=api_key or "not-required",  # Some local servers don't need API key
-                base_url=api_url,  # None = default OpenAI endpoint
-                timeout=float(timeout),
-            )
+            client_kwargs: dict[str, Any] = {
+                "api_key": api_key or "not-required",
+                "base_url": api_url,
+                "timeout": float(timeout),
+            }
+            if default_headers:
+                client_kwargs["default_headers"] = default_headers
+
+            client = AsyncOpenAI(**client_kwargs)
 
             response = await client.embeddings.create(
                 model=model,

src/workflows_mcp/engine/llm_config.py

@@ -117,6 +117,13 @@ class ProviderConfig(BaseModel):
         le=60.0,
         description="Initial retry delay in seconds (exponential backoff)",
     )
+    extra_headers: dict[str, str] = Field(
+        default_factory=dict,
+        description=(
+            "Custom HTTP headers to include in all requests to this provider. "
+            "Values can use {ENV_VAR} syntax for runtime environment variable resolution."
+        ),
+    )
     # Azure OpenAI specific fields
     deployment_name: str | None = Field(
         default=None,
@@ -237,6 +244,10 @@ class ResolvedLLMConfig(BaseModel):
     temperature: float | None = Field(default=None, description="Sampling temperature")
     max_tokens: int | None = Field(default=None, description="Maximum tokens to generate")
     system_instructions: str | None = Field(default=None, description="System instructions")
+    extra_headers: dict[str, str] = Field(
+        default_factory=dict,
+        description="Custom HTTP headers for provider requests",
+    )
     # Azure OpenAI specific
     deployment_name: str | None = Field(default=None)
     api_version: str | None = Field(default=None)
@@ -447,6 +458,7 @@ def resolve_profile(
             system_instructions=inline.get("system_instructions"),
             deployment_name=inline.get("deployment_name", provider_config.deployment_name),
             api_version=inline.get("api_version", provider_config.api_version),
+            extra_headers=inline.get("extra_headers", provider_config.extra_headers),
         )
 
     def get_default_profile(self) -> str | None: