522 lines
24 KiB
Python
522 lines
24 KiB
Python
"""Unified extensions configuration for MCP servers and skills."""
|
|
|
|
import copy
|
|
import errno
|
|
import json
|
|
import logging
|
|
import os
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
from collections.abc import Callable, Iterator
|
|
from contextlib import contextmanager
|
|
from pathlib import Path
|
|
from typing import Any, Literal
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
|
|
from deerflow.config.runtime_paths import existing_project_file
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_EXTENSIONS_CONFIG_WRITE_LOCK = threading.Lock()
|
|
|
|
|
|
class McpOAuthConfig(BaseModel):
|
|
"""OAuth configuration for an MCP server (HTTP/SSE transports)."""
|
|
|
|
enabled: bool = Field(default=True, description="Whether OAuth token injection is enabled")
|
|
token_url: str = Field(description="OAuth token endpoint URL")
|
|
grant_type: Literal["client_credentials", "refresh_token"] = Field(
|
|
default="client_credentials",
|
|
description="OAuth grant type",
|
|
)
|
|
client_id: str | None = Field(default=None, description="OAuth client ID")
|
|
client_secret: str | None = Field(default=None, description="OAuth client secret")
|
|
refresh_token: str | None = Field(default=None, description="OAuth refresh token (for refresh_token grant)")
|
|
scope: str | None = Field(default=None, description="OAuth scope")
|
|
audience: str | None = Field(default=None, description="OAuth audience (provider-specific)")
|
|
token_field: str = Field(default="access_token", description="Field name containing access token in token response")
|
|
token_type_field: str = Field(default="token_type", description="Field name containing token type in token response")
|
|
expires_in_field: str = Field(default="expires_in", description="Field name containing expiry (seconds) in token response")
|
|
default_token_type: str = Field(default="Bearer", description="Default token type when missing in token response")
|
|
refresh_skew_seconds: int = Field(default=60, description="Refresh token this many seconds before expiry")
|
|
extra_token_params: dict[str, str] = Field(default_factory=dict, description="Additional form params sent to token endpoint")
|
|
model_config = ConfigDict(extra="allow")
|
|
|
|
|
|
class McpServerConfig(BaseModel):
|
|
"""Configuration for a single MCP server."""
|
|
|
|
enabled: bool = Field(default=True, description="Whether this MCP server is enabled")
|
|
type: str = Field(default="stdio", description="Transport type: 'stdio', 'sse', or 'http'")
|
|
command: str | None = Field(default=None, description="Command to execute to start the MCP server (for stdio type)")
|
|
args: list[str] = Field(default_factory=list, description="Arguments to pass to the command (for stdio type)")
|
|
env: dict[str, str] = Field(default_factory=dict, description="Environment variables for the MCP server")
|
|
url: str | None = Field(default=None, description="URL of the MCP server (for sse or http type)")
|
|
headers: dict[str, str] = Field(default_factory=dict, description="HTTP headers to send (for sse or http type)")
|
|
oauth: McpOAuthConfig | None = Field(default=None, description="OAuth configuration (for sse or http type)")
|
|
description: str = Field(default="", description="Human-readable description of what this MCP server provides")
|
|
disabled_tools: list[str] = Field(
|
|
default_factory=list,
|
|
description="List of tool names (without server-name prefix) to hide from the agent",
|
|
alias="disabledTools",
|
|
)
|
|
model_config = ConfigDict(extra="allow", populate_by_name=True)
|
|
|
|
|
|
class SkillDisplayTableColumn(BaseModel):
|
|
"""A configured table column for structured skill result display."""
|
|
|
|
label: str = Field(description="Column label shown in the frontend table")
|
|
field: str = Field(description="Field name or dot path read from each result item")
|
|
width: str | None = Field(default=None, description="Optional CSS width for this column")
|
|
model_config = ConfigDict(extra="allow", populate_by_name=True)
|
|
|
|
|
|
class SkillDisplayResultSetConfig(BaseModel):
|
|
"""Display adapter for one result array inside a skill response."""
|
|
|
|
key: str = Field(description="Stable key for this result set")
|
|
label: str | None = Field(default=None, description="Human-readable result set label")
|
|
enabled: bool = Field(default=True, description="Whether this result set is enabled")
|
|
mode: Literal["citation", "list", "table", "card", "custom", "none"] = Field(
|
|
default="citation",
|
|
description="How the frontend should render this result set",
|
|
)
|
|
result_path: str = Field(default="results", description="Dot path to the array of result items", alias="resultPath")
|
|
title_field: str | None = Field(default=None, description="Field used as item title", alias="titleField")
|
|
summary_field: str | None = Field(default=None, description="Field used as short summary/snippet", alias="summaryField")
|
|
content_field: str | None = Field(default=None, description="Field used as full item content", alias="contentField")
|
|
source_field: str | None = Field(default=None, description="Field used as source label", alias="sourceField")
|
|
author_field: str | None = Field(default=None, description="Field used as author/byline label", alias="authorField")
|
|
type_field: str | None = Field(default=None, description="Field used as source/type badge", alias="typeField")
|
|
time_field: str | None = Field(default=None, description="Field used as publication/update time", alias="timeField")
|
|
score_field: str | None = Field(default=None, description="Field used as relevance score", alias="scoreField")
|
|
url_field: str | None = Field(default=None, description="Field containing a direct URL", alias="urlField")
|
|
id_field: str | None = Field(default=None, description="Field containing an ID for template URLs", alias="idField")
|
|
link_strategy: Literal["auto", "url", "id"] = Field(
|
|
default="auto",
|
|
description="Whether links use a URL field, an ID template, or automatic fallback",
|
|
alias="linkStrategy",
|
|
)
|
|
link_template: str | None = Field(default=None, description="Template used when linkStrategy is id", alias="linkTemplate")
|
|
citation_token: Literal["doubleBracket", "singleBracket"] = Field(
|
|
default="singleBracket",
|
|
description="Citation token style expected in model output",
|
|
alias="citationToken",
|
|
)
|
|
max_items: int = Field(default=100, ge=1, le=100, description="Maximum displayed items", alias="maxItems")
|
|
custom_instruction: str = Field(
|
|
default="",
|
|
description="User-authored instruction for custom rendering schema generation",
|
|
alias="customInstruction",
|
|
)
|
|
table_columns: list[SkillDisplayTableColumn] = Field(
|
|
default_factory=list,
|
|
description="Columns used when mode is table",
|
|
alias="tableColumns",
|
|
)
|
|
model_config = ConfigDict(extra="allow", populate_by_name=True)
|
|
|
|
|
|
class SkillDisplayConfig(BaseModel):
|
|
"""Frontend display adapter for structured skill results."""
|
|
|
|
enabled: bool = Field(default=False, description="Whether result display adaptation is enabled")
|
|
mode: Literal["citation", "list", "table", "card", "custom", "none"] = Field(
|
|
default="none",
|
|
description="How the frontend should render this skill's structured results",
|
|
)
|
|
result_path: str = Field(default="results", description="Dot path to the array of result items", alias="resultPath")
|
|
title_field: str | None = Field(default=None, description="Field used as item title", alias="titleField")
|
|
summary_field: str | None = Field(default=None, description="Field used as short summary/snippet", alias="summaryField")
|
|
content_field: str | None = Field(default=None, description="Field used as full item content", alias="contentField")
|
|
source_field: str | None = Field(default=None, description="Field used as source label", alias="sourceField")
|
|
author_field: str | None = Field(default=None, description="Field used as author/byline label", alias="authorField")
|
|
type_field: str | None = Field(default=None, description="Field used as source/type badge", alias="typeField")
|
|
time_field: str | None = Field(default=None, description="Field used as publication/update time", alias="timeField")
|
|
score_field: str | None = Field(default=None, description="Field used as relevance score", alias="scoreField")
|
|
url_field: str | None = Field(default=None, description="Field containing a direct URL", alias="urlField")
|
|
id_field: str | None = Field(default=None, description="Field containing an ID for template URLs", alias="idField")
|
|
link_strategy: Literal["auto", "url", "id"] = Field(
|
|
default="auto",
|
|
description="Whether links use a URL field, an ID template, or automatic fallback",
|
|
alias="linkStrategy",
|
|
)
|
|
link_template: str | None = Field(default=None, description="Template used when linkStrategy is id", alias="linkTemplate")
|
|
citation_token: Literal["doubleBracket", "singleBracket"] = Field(
|
|
default="doubleBracket",
|
|
description="Citation token style expected in model output",
|
|
alias="citationToken",
|
|
)
|
|
max_items: int = Field(default=8, ge=1, le=100, description="Maximum displayed items", alias="maxItems")
|
|
custom_instruction: str = Field(
|
|
default="",
|
|
description="User-authored instruction for custom rendering schema generation",
|
|
alias="customInstruction",
|
|
)
|
|
table_columns: list[SkillDisplayTableColumn] = Field(
|
|
default_factory=list,
|
|
description="Columns used when mode is table",
|
|
alias="tableColumns",
|
|
)
|
|
result_sets: list[SkillDisplayResultSetConfig] = Field(
|
|
default_factory=list,
|
|
description="Optional per-result-array display adapters",
|
|
alias="resultSets",
|
|
)
|
|
model_config = ConfigDict(extra="allow", populate_by_name=True)
|
|
|
|
|
|
class SkillStateConfig(BaseModel):
|
|
"""Configuration for a single skill's state."""
|
|
|
|
enabled: bool = Field(default=True, description="Whether this skill is enabled")
|
|
display: SkillDisplayConfig | None = Field(default=None, description="Optional frontend display adapter")
|
|
|
|
|
|
class ExtensionsConfig(BaseModel):
|
|
"""Unified configuration for MCP servers and skills."""
|
|
|
|
mcp_servers: dict[str, McpServerConfig] = Field(
|
|
default_factory=dict,
|
|
description="Map of MCP server name to configuration",
|
|
alias="mcpServers",
|
|
)
|
|
skills: dict[str, SkillStateConfig] = Field(
|
|
default_factory=dict,
|
|
description="Map of skill name to state configuration",
|
|
)
|
|
model_config = ConfigDict(extra="allow", populate_by_name=True)
|
|
|
|
@classmethod
|
|
def resolve_config_path(cls, config_path: str | None = None) -> Path | None:
|
|
"""Resolve the extensions config file path.
|
|
|
|
Priority:
|
|
1. If provided `config_path` argument, use it.
|
|
2. If provided `DEER_FLOW_EXTENSIONS_CONFIG_PATH` environment variable, use it.
|
|
3. Otherwise, search the caller project root for `extensions_config.json`, then `mcp_config.json`.
|
|
4. For backward compatibility, also search legacy backend/repository-root defaults.
|
|
5. If not found, return None (extensions are optional).
|
|
|
|
Args:
|
|
config_path: Optional path to extensions config file.
|
|
|
|
Resolution order:
|
|
1. If provided `config_path` argument, use it.
|
|
2. If provided `DEER_FLOW_EXTENSIONS_CONFIG_PATH` environment variable, use it.
|
|
3. Otherwise, search the caller project root for
|
|
`extensions_config.json`, then legacy `mcp_config.json`.
|
|
4. Finally, search backend/repository-root defaults for monorepo compatibility.
|
|
|
|
Returns:
|
|
Path to the extensions config file if found, otherwise None.
|
|
"""
|
|
if config_path:
|
|
path = Path(config_path)
|
|
if not path.exists():
|
|
raise FileNotFoundError(f"Extensions config file specified by param `config_path` not found at {path}")
|
|
return path
|
|
elif os.getenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH"):
|
|
path = Path(os.getenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH"))
|
|
if not path.exists():
|
|
raise FileNotFoundError(f"Extensions config file specified by environment variable `DEER_FLOW_EXTENSIONS_CONFIG_PATH` not found at {path}")
|
|
return path
|
|
else:
|
|
project_config = existing_project_file(("extensions_config.json", "mcp_config.json"))
|
|
if project_config is not None:
|
|
return project_config
|
|
|
|
backend_dir = Path(__file__).resolve().parents[4]
|
|
repo_root = backend_dir.parent
|
|
for path in (
|
|
backend_dir / "extensions_config.json",
|
|
repo_root / "extensions_config.json",
|
|
backend_dir / "mcp_config.json",
|
|
repo_root / "mcp_config.json",
|
|
):
|
|
if path.exists():
|
|
return path
|
|
|
|
# Extensions are optional, so return None if not found
|
|
return None
|
|
|
|
@classmethod
|
|
def from_file(cls, config_path: str | None = None) -> "ExtensionsConfig":
|
|
"""Load extensions config from JSON file.
|
|
|
|
See `resolve_config_path` for more details.
|
|
|
|
Args:
|
|
config_path: Path to the extensions config file.
|
|
|
|
Returns:
|
|
ExtensionsConfig: The loaded config, or empty config if file not found.
|
|
"""
|
|
resolved_path = cls.resolve_config_path(config_path)
|
|
if resolved_path is None:
|
|
# Return empty config if extensions config file is not found
|
|
return cls(mcp_servers={}, skills={})
|
|
|
|
try:
|
|
with open(resolved_path, encoding="utf-8") as f:
|
|
config_data = json.load(f)
|
|
cls.resolve_env_variables(config_data)
|
|
return cls.model_validate(config_data)
|
|
except json.JSONDecodeError as e:
|
|
raise ValueError(f"Extensions config file at {resolved_path} is not valid JSON: {e}") from e
|
|
except Exception as e:
|
|
raise RuntimeError(f"Failed to load extensions config from {resolved_path}: {e}") from e
|
|
|
|
@classmethod
|
|
def resolve_env_variables(cls, config: dict[str, Any]) -> dict[str, Any]:
|
|
"""Recursively resolve environment variables in the config.
|
|
|
|
Environment variables are resolved using the `os.getenv` function. Example: $OPENAI_API_KEY
|
|
|
|
Args:
|
|
config: The config to resolve environment variables in.
|
|
|
|
Returns:
|
|
The config with environment variables resolved.
|
|
"""
|
|
for key, value in config.items():
|
|
if isinstance(value, str):
|
|
if value.startswith("$"):
|
|
env_value = os.getenv(value[1:])
|
|
if env_value is None:
|
|
# Unresolved placeholder — store empty string so downstream
|
|
# consumers (e.g. MCP servers) don't receive the literal "$VAR"
|
|
# token as an actual environment value.
|
|
config[key] = ""
|
|
else:
|
|
config[key] = env_value
|
|
else:
|
|
config[key] = value
|
|
elif isinstance(value, dict):
|
|
config[key] = cls.resolve_env_variables(value)
|
|
elif isinstance(value, list):
|
|
config[key] = [cls.resolve_env_variables(item) if isinstance(item, dict) else item for item in value]
|
|
return config
|
|
|
|
def get_enabled_mcp_servers(self) -> dict[str, McpServerConfig]:
|
|
"""Get only the enabled MCP servers.
|
|
|
|
Returns:
|
|
Dictionary of enabled MCP servers.
|
|
"""
|
|
return {name: config for name, config in self.mcp_servers.items() if config.enabled}
|
|
|
|
def is_skill_enabled(self, skill_name: str, skill_category: str) -> bool:
|
|
"""Check if a skill is enabled.
|
|
|
|
Args:
|
|
skill_name: Name of the skill
|
|
skill_category: Category of the skill
|
|
|
|
Returns:
|
|
True if enabled, False otherwise
|
|
"""
|
|
skill_config = self.skills.get(skill_name)
|
|
if skill_config is None:
|
|
# Default to enable for public & custom skill
|
|
return skill_category in ("public", "custom")
|
|
return skill_config.enabled
|
|
|
|
|
|
_extensions_config: ExtensionsConfig | None = None
|
|
|
|
|
|
def get_extensions_config() -> ExtensionsConfig:
|
|
"""Get the extensions config instance.
|
|
|
|
Returns a cached singleton instance. Use `reload_extensions_config()` to reload
|
|
from file, or `reset_extensions_config()` to clear the cache.
|
|
|
|
Returns:
|
|
The cached ExtensionsConfig instance.
|
|
"""
|
|
global _extensions_config
|
|
if _extensions_config is None:
|
|
_extensions_config = ExtensionsConfig.from_file()
|
|
return _extensions_config
|
|
|
|
|
|
def reload_extensions_config(config_path: str | None = None) -> ExtensionsConfig:
|
|
"""Reload the extensions config from file and update the cached instance.
|
|
|
|
This is useful when the config file has been modified and you want
|
|
to pick up the changes without restarting the application.
|
|
|
|
Args:
|
|
config_path: Optional path to extensions config file. If not provided,
|
|
uses the default resolution strategy.
|
|
|
|
Returns:
|
|
The newly loaded ExtensionsConfig instance.
|
|
"""
|
|
global _extensions_config
|
|
_extensions_config = ExtensionsConfig.from_file(config_path)
|
|
return _extensions_config
|
|
|
|
|
|
def _read_raw_extensions_config(config_path: Path) -> dict[str, Any]:
|
|
"""Read the raw JSON document without resolving env placeholders."""
|
|
if not config_path.exists():
|
|
return {}
|
|
with config_path.open(encoding="utf-8") as config_file:
|
|
document = json.load(config_file)
|
|
if not isinstance(document, dict):
|
|
raise ValueError("Extensions config root must be a JSON object")
|
|
return document
|
|
|
|
|
|
def read_extensions_config_document(config_path: str | Path) -> dict[str, Any]:
|
|
"""Return a fresh raw document while coordinating with all writers."""
|
|
path = Path(config_path)
|
|
with _EXTENSIONS_CONFIG_WRITE_LOCK:
|
|
with _extensions_config_process_lock(path):
|
|
return _read_raw_extensions_config(path)
|
|
|
|
|
|
@contextmanager
|
|
def _extensions_config_process_lock(config_path: Path) -> Iterator[None]:
|
|
"""Serialize writers across processes on Windows and POSIX.
|
|
|
|
The stable sidecar is intentional: deleting it after unlock can let two
|
|
processes lock different inodes. Keeping it beside the config also lets
|
|
multiple processes sharing the writable config directory coordinate.
|
|
"""
|
|
lock_path = config_path.with_name(f".{config_path.name}.lock")
|
|
lock_path.parent.mkdir(parents=True, exist_ok=True)
|
|
with lock_path.open("a+b") as lock_file:
|
|
if os.name == "nt":
|
|
import msvcrt
|
|
|
|
lock_file.seek(0, os.SEEK_END)
|
|
if lock_file.tell() == 0:
|
|
lock_file.write(b"\0")
|
|
lock_file.flush()
|
|
lock_file.seek(0)
|
|
while True:
|
|
try:
|
|
msvcrt.locking(lock_file.fileno(), msvcrt.LK_NBLCK, 1)
|
|
break
|
|
except OSError as exc:
|
|
if exc.errno not in {errno.EACCES, errno.EAGAIN, errno.EDEADLK}:
|
|
raise
|
|
time.sleep(0.05)
|
|
try:
|
|
yield
|
|
finally:
|
|
lock_file.seek(0)
|
|
msvcrt.locking(lock_file.fileno(), msvcrt.LK_UNLCK, 1)
|
|
else:
|
|
import fcntl
|
|
|
|
fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX)
|
|
try:
|
|
yield
|
|
finally:
|
|
fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)
|
|
|
|
|
|
def _write_raw_extensions_config(config_path: Path, document: dict[str, Any]) -> None:
|
|
"""Durably replace raw JSON, with a bind-mounted-file EBUSY fallback."""
|
|
config_path.parent.mkdir(parents=True, exist_ok=True)
|
|
payload = json.dumps(document, indent=2, ensure_ascii=False) + "\n"
|
|
tmp_path: Path | None = None
|
|
try:
|
|
with tempfile.NamedTemporaryFile(
|
|
"w",
|
|
encoding="utf-8",
|
|
delete=False,
|
|
dir=str(config_path.parent),
|
|
) as tmp_file:
|
|
tmp_file.write(payload)
|
|
tmp_file.flush()
|
|
os.fsync(tmp_file.fileno())
|
|
tmp_path = Path(tmp_file.name)
|
|
try:
|
|
os.replace(tmp_path, config_path)
|
|
except OSError as exc:
|
|
if exc.errno != errno.EBUSY:
|
|
raise
|
|
logger.warning(
|
|
"Atomic replace is unavailable for bind-mounted extensions config %s; using in-place fallback",
|
|
config_path,
|
|
)
|
|
previous = config_path.read_bytes() if config_path.exists() else None
|
|
try:
|
|
mode = "r+" if config_path.exists() else "w"
|
|
with config_path.open(mode, encoding="utf-8") as config_file:
|
|
config_file.seek(0)
|
|
config_file.write(payload)
|
|
config_file.truncate()
|
|
config_file.flush()
|
|
os.fsync(config_file.fileno())
|
|
except Exception:
|
|
if previous is not None:
|
|
try:
|
|
with config_path.open("wb") as config_file:
|
|
config_file.write(previous)
|
|
config_file.flush()
|
|
os.fsync(config_file.fileno())
|
|
except Exception:
|
|
logger.exception("Failed to restore extensions config after in-place write failure")
|
|
raise
|
|
finally:
|
|
if tmp_path is not None and tmp_path.exists():
|
|
tmp_path.unlink()
|
|
|
|
|
|
def patch_extensions_config(
|
|
config_path: str | Path,
|
|
patch: Callable[[dict[str, Any]], None],
|
|
) -> ExtensionsConfig:
|
|
"""Apply a raw-document patch under thread and cross-process locks.
|
|
|
|
Callers should mutate only the fields they own. The document is read fresh
|
|
while locked, so concurrent MCP, skill-state, and display writes cannot
|
|
overwrite one another or resolve/persist secret ``$ENV`` placeholders.
|
|
"""
|
|
path = Path(config_path)
|
|
with _EXTENSIONS_CONFIG_WRITE_LOCK:
|
|
with _extensions_config_process_lock(path):
|
|
document = _read_raw_extensions_config(path)
|
|
patch(document)
|
|
if not isinstance(document, dict):
|
|
raise ValueError("Extensions config root must be a JSON object")
|
|
# Validate a deep copy before persisting. Environment expansion is
|
|
# intentionally confined to the copy so placeholders remain raw.
|
|
validation_document = copy.deepcopy(document)
|
|
ExtensionsConfig.resolve_env_variables(validation_document)
|
|
ExtensionsConfig.model_validate(validation_document)
|
|
_write_raw_extensions_config(path, document)
|
|
return reload_extensions_config(str(path))
|
|
|
|
|
|
def reset_extensions_config() -> None:
|
|
"""Reset the cached extensions config instance.
|
|
|
|
This clears the singleton cache, causing the next call to
|
|
`get_extensions_config()` to reload from file. Useful for testing
|
|
or when switching between different configurations.
|
|
"""
|
|
global _extensions_config
|
|
_extensions_config = None
|
|
|
|
|
|
def set_extensions_config(config: ExtensionsConfig) -> None:
|
|
"""Set a custom extensions config instance.
|
|
|
|
This allows injecting a custom or mock config for testing purposes.
|
|
|
|
Args:
|
|
config: The ExtensionsConfig instance to use.
|
|
"""
|
|
global _extensions_config
|
|
_extensions_config = config
|