From fc2ab7796a8cc7e8c6b15c16585fbd57e982c2a0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marcelo=20M=C3=B3dolo?= Date: Sun, 22 Mar 2026 17:39:12 -0300 Subject: [PATCH] feat: support ${VAR} environment variable interpolation in headers Expand ${VAR_NAME} references in X-Target-URL, X-Target-Headers, and X-Base-URL header values using os.environ at request time. This allows MCP client configurations to be version-controlled without embedding secrets or environment-specific values directly: "X-Target-Headers": "{\"Authorization\": \"Bearer ${MY_TOKEN\"}" Unknown variables are left as-is for easy debugging of incomplete configs. --- api_agent/context.py | 18 +++++++++++++++--- 1 file changed, 15 insertions(+), 3 deletions(-) diff --git a/api_agent/context.py b/api_agent/context.py index 611732c..17b3b39 100644 --- a/api_agent/context.py +++ b/api_agent/context.py @@ -1,6 +1,7 @@ """Request context extraction from HTTP headers.""" import json +import os import re from dataclasses import dataclass from urllib.parse import urlparse @@ -8,6 +9,17 @@ from fastmcp.server.dependencies import get_http_headers +_ENV_VAR_RE = re.compile(r"\$\{([^}]+)\}") + + +def _expand_env_vars(value: str) -> str: + """Expand ``${VAR}`` references in *value* using environment variables. + + Unknown variables are left as-is. + """ + return _ENV_VAR_RE.sub(lambda m: os.environ.get(m.group(1), m.group(0)), value) + + class MissingHeaderError(Exception): """Required header missing from request.""" @@ -46,15 +58,15 @@ def get_request_context() -> RequestContext: """ headers = get_http_headers() - target_url = headers.get("x-target-url") + target_url = _expand_env_vars(headers.get("x-target-url") or "") or None api_type = headers.get("x-api-type") - target_headers_raw = headers.get("x-target-headers") or "{}" + target_headers_raw = _expand_env_vars(headers.get("x-target-headers") or "{}") allow_unsafe_paths_raw = headers.get("x-allow-unsafe-paths") or "[]" base_url_raw = headers.get("x-base-url") include_result_raw = headers.get("x-include-result", "false") poll_paths_raw = headers.get("x-poll-paths") or "[]" - base_url = base_url_raw if base_url_raw else None + base_url = _expand_env_vars(base_url_raw) if base_url_raw else None include_result = (include_result_raw or "").lower() in ("true", "1", "yes") if not target_url: