portugalfuturista/replica-omnisciente
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 72

feat(mcp-router): load provider registry from mcp.providers.json mirror

Browse source 
model_router.py now builds its runtime REGISTRY from the generated
.aurelio/providers/dist/mcp.providers.json (AURELIO_MCP_PROVIDERS_JSON to
override), expanding the Provider enum to the full Hermes parity set and
dispatching chat by protocol (one OpenAI-compatible path covers ~11 providers;
Anthropic/Gemini/Bedrock/portal/subprocess native). Hard-coded registry is
retained only as a fallback when the mirror is absent.
This commit is contained in:
fabiorafaelcoutada 2026-07-10 22:39:18 +01:00
parent dff3ecebc8
commit 23782cbfb4
1 changed files with 222 additions and 85 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

307
.aurelio/mcp/electrical-mcp/servers/mcp-knowledge-server/src/intelligence/model_router.py
View file

@ -1,11 +1,14 @@
"""Multi-Provider Model Router.
Routes inference and embedding requests across all available LLM providers.
Supports: OpenAI, Google Gemini, Anthropic Claude, Kimi (Moonshot).
The provider list is loaded from the generated mirror
``.aurelio/providers/dist/mcp.providers.json`` (single source of truth:
``.aurelio/providers/registry.yaml``). If that file is absent — e.g. the MCP
server is deployed standalone — a built-in fallback registry is used instead.
Provider selection strategy:
1. Explicit provider override (user asks for a specific model).
2. Task-based routing (deep reasoning -> Claude/Gemini, fast lookup -> Kimi/OpenAI).
2. Task-based routing (deep reasoning -> Claude/Gemini, fast lookup -> Kimi/Qwen).
3. Cascading fallback (if primary provider fails, try next available).
All API keys are read from environment variables at init time.
@ -16,6 +19,7 @@ import json
import asyncio
from enum import Enum
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional
import httpx
@ -24,12 +28,51 @@ import structlog
logger = structlog.get_logger()
# Path to the generated registry mirror. Resolved relative to the repo root
# (replica-omnisciente/.aurelio/providers/dist/mcp.providers.json). The MCP
# package lives deep under .aurelio/mcp/..., so we walk up to find .aurelio.
def _locate_mcp_registry() -> Optional[Path]:
here = Path(__file__).resolve()
for parent in here.parents:
candidate = parent / "providers" / "dist" / "mcp.providers.json"
if candidate.exists():
return candidate
# Env override for deployments that mount the mirror elsewhere.
env = os.environ.get("AURELIO_MCP_PROVIDERS_JSON")
if env:
p = Path(env)
if p.exists():
return p
return None
class Provider(str, Enum):
# Core / classic
OPENAI = "openai"
GEMINI = "gemini"
CLAUDE = "claude"
KIMI = "kimi"
LOCAL = "local" # sentence-transformers, no API key needed
CLAUDE = "claude" # kept as alias; enum value is "claude" for back-compat
ANTHROPIC = "anthropic"
KIMI = "kimi" # legacy alias -> kimi-coding
MOONSHOT = "moonshot"
LOCAL = "local" # sentence-transformers / Ollama, no API key needed
# Hermes-parity
OPENROUTER = "openrouter"
OPENAI_CODEX = "openai-codex"
NOUS = "nous"
ZAI = "zai"
KIMI_CODING = "kimi-coding"
KIMI_CODING_CN = "kimi-coding-cn"
MINIMAX = "minimax"
MINIMAX_CN = "minimax-cn"
BEDROCK = "bedrock"
# Alibaba Cloud
QWEN = "qwen"
QWEN_CODE = "qwen-code"
ALIBABA_CLOUD = "alibaba-cloud"
# CLI subprocess / portal
ANTIGRAVITY = "antigravity"
CLAUDE_CODE = "claude-code"
MIMOCODE = "mimocode"
@dataclass
@ -40,72 +83,131 @@ class ProviderConfig:
chat_model: str
embedding_model: Optional[str]
max_context: int # tokens
protocol: str = "openai" # openai | anthropic | gemini | bedrock-converse | portal | subprocess
auth: str = "api_key" # api_key | oauth | iam | none
key_env: Optional[str] = None
is_available: bool = field(init=False)
def __post_init__(self):
self.is_available = bool(self.api_key)
# local / none-auth providers are "available" even with no key;
# oauth providers are considered available (login handled out-of-band).
if self.auth in ("none", "oauth"):
self.is_available = True
else:
self.is_available = bool(self.api_key)
# ---------------------------------------------------------------------------
# Provider registry — built once at import time from env vars
# Provider registry — loaded from the generated mcp.providers.json mirror
# (single source of truth: .aurelio/providers/registry.yaml). Falls back to a
# small built-in set when the mirror is absent (standalone MCP deployments).
# ---------------------------------------------------------------------------
def _build_registry() -> dict[Provider, ProviderConfig]:
# Map protocol strings to the env var / dispatch used. Anthropic and Gemini keep
# their native protocols; everything OpenAI-compatible shares `_chat_openai_compat`.
# Legacy alias: callers that pass Provider.CLAUDE / Provider.KIMI still resolve.
_ALIAS = {
Provider.CLAUDE: Provider.ANTHROPIC,
Provider.KIMI: Provider.KIMI_CODING,
}
def _builtin_registry() -> dict[Provider, ProviderConfig]:
"""Minimal fallback when mcp.providers.json is not present."""
return {
Provider.OPENAI: ProviderConfig(
name=Provider.OPENAI,
api_key=os.environ.get("OPENAI_API_KEY", ""),
base_url="https://api.openai.com/v1",
chat_model="gpt-4o",
embedding_model="text-embedding-3-large",
max_context=128_000,
),
name=Provider.OPENAI, api_key=os.environ.get("OPENAI_API_KEY", ""),
base_url="https://api.openai.com/v1", chat_model="gpt-4o",
embedding_model="text-embedding-3-large", max_context=128_000,
protocol="openai", auth="api_key", key_env="OPENAI_API_KEY"),
Provider.GEMINI: ProviderConfig(
name=Provider.GEMINI,
api_key=os.environ.get("GEMINI_API_KEY", ""),
base_url="https://generativelanguage.googleapis.com/v1beta",
chat_model="gemini-2.5-pro",
embedding_model="text-embedding-004",
max_context=1_000_000,
),
Provider.CLAUDE: ProviderConfig(
name=Provider.CLAUDE,
api_key=os.environ.get("ANTHROPIC_API_KEY", ""),
base_url="https://api.anthropic.com/v1",
chat_model="claude-sonnet-4-20250514",
embedding_model=None, # Claude has no embedding endpoint
max_context=200_000,
),
Provider.KIMI: ProviderConfig(
name=Provider.KIMI,
api_key=os.environ.get("KIMI_API_KEY", ""),
base_url="https://api.moonshot.ai/v1",
chat_model="moonshot-v1-128k",
embedding_model=None, # Kimi has no public embedding endpoint
max_context=128_000,
),
name=Provider.GEMINI, api_key=os.environ.get("GEMINI_API_KEY", ""),
base_url="https://generativelanguage.googleapis.com/v1beta", chat_model="gemini-2.5-pro",
embedding_model="text-embedding-004", max_context=1_000_000,
protocol="gemini", auth="api_key", key_env="GEMINI_API_KEY"),
Provider.ANTHROPIC: ProviderConfig(
name=Provider.ANTHROPIC, api_key=os.environ.get("ANTHROPIC_API_KEY", ""),
base_url="https://api.anthropic.com/v1", chat_model="claude-sonnet-4-20250514",
embedding_model=None, max_context=200_000,
protocol="anthropic", auth="api_key", key_env="ANTHROPIC_API_KEY"),
Provider.KIMI_CODING: ProviderConfig(
name=Provider.KIMI_CODING, api_key=os.environ.get("KIMI_API_KEY", ""),
base_url="https://api.moonshot.ai/v1", chat_model="kimi-k2-0905-preview",
embedding_model=None, max_context=131_072,
protocol="openai", auth="api_key", key_env="KIMI_API_KEY"),
Provider.LOCAL: ProviderConfig(
name=Provider.LOCAL,
api_key="not-needed",
base_url="",
chat_model="",
embedding_model="all-MiniLM-L6-v2",
max_context=512,
),
name=Provider.LOCAL, api_key="not-needed", base_url="http://127.0.0.1:11434",
chat_model="qwen2.5-coder:14b", embedding_model="all-MiniLM-L6-v2", max_context=32_768,
protocol="openai", auth="none", key_env=None),
}
REGISTRY: dict[Provider, ProviderConfig] = _build_registry()
def _load_registry_from_mirror(path: Path) -> tuple[dict[Provider, ProviderConfig], list[Provider], list[Provider]]:
data = json.loads(path.read_text(encoding="utf-8"))
registry: dict[Provider, ProviderConfig] = {}
for entry in data.get("providers", []):
try:
prov = Provider(entry["id"])
except ValueError:
logger.warning("model_router.unknown_provider", id=entry.get("id"))
continue
key_env = entry.get("keyEnv")
api_key = os.environ.get(key_env, "") if key_env else ("not-needed" if entry.get("auth") in ("none", "oauth") else "")
registry[prov] = ProviderConfig(
name=prov,
api_key=api_key,
base_url=entry.get("baseUrl") or "",
chat_model=entry.get("chatModel") or "",
embedding_model=entry.get("embeddingModel"),
max_context=int(entry.get("maxContext") or 0),
protocol=entry.get("protocol", "openai"),
auth=entry.get("auth", "api_key"),
key_env=key_env,
)
chat_fb = [Provider(p) for p in data.get("chatFallback", []) if p in Provider._value2member_map_]
embed_fb = [Provider(p) for p in data.get("embedFallback", []) if p in Provider._value2member_map_]
return registry, chat_fb, embed_fb
_MIRROR = _locate_mcp_registry()
if _MIRROR is not None:
try:
REGISTRY, _CHAT_PREFERENCE, _EMBED_PREFERENCE = _load_registry_from_mirror(_MIRROR)
logger.info("model_router.registry_loaded", path=str(_MIRROR), providers=len(REGISTRY))
except Exception as exc: # pragma: no cover — defensive
logger.warning("model_router.registry_load_failed", error=str(exc))
REGISTRY = _builtin_registry()
_CHAT_PREFERENCE = [Provider.KIMI_CODING, Provider.GEMINI, Provider.ANTHROPIC, Provider.OPENAI]
_EMBED_PREFERENCE = [Provider.GEMINI, Provider.OPENAI]
else:
REGISTRY = _builtin_registry()
_CHAT_PREFERENCE = [Provider.KIMI_CODING, Provider.GEMINI, Provider.ANTHROPIC, Provider.OPENAI]
_EMBED_PREFERENCE = [Provider.GEMINI, Provider.OPENAI]
def resolve_provider(provider: Provider) -> Provider:
"""Map legacy aliases (CLAUDE, KIMI) to their canonical provider."""
return _ALIAS.get(provider, provider)
def available_providers() -> list[Provider]:
"""Return providers that have a valid API key configured."""
"""Return providers that have a valid API key configured (or are keyless/oauth)."""
return [p for p, cfg in REGISTRY.items() if cfg.is_available]
def available_chat_providers() -> list[Provider]:
"""Providers capable of chat/completion (excludes LOCAL)."""
return [p for p in available_providers() if p != Provider.LOCAL and REGISTRY[p].chat_model]
"""Providers capable of chat/completion (excludes LOCAL and subprocess-only)."""
out = []
for p in available_providers():
cfg = REGISTRY[p]
if p == Provider.LOCAL:
continue
if cfg.protocol == "subprocess":
continue
if cfg.chat_model:
out.append(p)
return out
def available_embedding_providers() -> list[Provider]:
@ -118,10 +220,16 @@ def available_embedding_providers() -> list[Provider]:
# ---------------------------------------------------------------------------
async def _chat_openai(cfg: ProviderConfig, system: str, prompt: str) -> str:
"""OpenAI-compatible chat/completions. Used by openai, kimi-coding, qwen,
qwen-code, minimax, minimax-cn, zai, openrouter, moonshot, alibaba-cloud,
nous, openai-codex — every provider whose protocol == 'openai'."""
headers = {"Authorization": f"Bearer {cfg.api_key}"} if cfg.api_key and cfg.auth == "api_key" else {}
if cfg.api_key and cfg.auth != "api_key":
headers["Authorization"] = f"Bearer {cfg.api_key}"
async with httpx.AsyncClient(timeout=60.0) as client:
resp = await client.post(
f"{cfg.base_url}/chat/completions",
headers={"Authorization": f"Bearer {cfg.api_key}"},
headers=headers,
json={
"model": cfg.chat_model,
"messages": [
@ -151,7 +259,7 @@ async def _chat_gemini(cfg: ProviderConfig, system: str, prompt: str) -> str:
return data["candidates"][0]["content"]["parts"][0]["text"]
async def _chat_claude(cfg: ProviderConfig, system: str, prompt: str) -> str:
async def _chat_anthropic(cfg: ProviderConfig, system: str, prompt: str) -> str:
async with httpx.AsyncClient(timeout=90.0) as client:
resp = await client.post(
f"{cfg.base_url}/messages",
@ -171,30 +279,48 @@ async def _chat_claude(cfg: ProviderConfig, system: str, prompt: str) -> str:
return resp.json()["content"][0]["text"]
async def _chat_kimi(cfg: ProviderConfig, system: str, prompt: str) -> str:
"""Kimi (Moonshot) uses an OpenAI-compatible API."""
async with httpx.AsyncClient(timeout=60.0) as client:
async def _chat_bedrock(cfg: ProviderConfig, system: str, prompt: str) -> str:
"""AWS Bedrock via the Converse API (boto3 / IAM). Requires boto3 + creds."""
import boto3 # type: ignore
region = os.environ.get("AWS_REGION", os.environ.get("AWS_DEFAULT_REGION", "us-east-1"))
client = boto3.client("bedrock-runtime", region_name=region)
body = {
"messages": [{"role": "user", "content": [{"text": prompt}]}],
"system": [{"text": system}],
"inferenceConfig": {"maxTokens": 4096, "temperature": 0.4},
}
loop = asyncio.get_running_loop()
resp = await loop.run_in_executor(
None,
lambda: client.converse(modelId=cfg.chat_model, **body),
)
return resp["output"]["message"]["content"][0]["text"]
async def _chat_portal(cfg: ProviderConfig, system: str, prompt: str) -> str:
"""Relay through a portugalfuturista.org-hosted agent (e.g. antigravity)."""
headers = {}
if cfg.api_key:
headers["Authorization"] = f"Bearer {cfg.api_key}"
async with httpx.AsyncClient(timeout=120.0) as client:
resp = await client.post(
f"{cfg.base_url}/chat/completions",
headers={"Authorization": f"Bearer {cfg.api_key}"},
json={
"model": cfg.chat_model,
"messages": [
{"role": "system", "content": system},
{"role": "user", "content": prompt},
],
"temperature": 0.4,
},
f"{cfg.base_url}/api/prompt",
headers=headers,
json={"system": system, "prompt": prompt},
)
resp.raise_for_status()
return resp.json()["choices"][0]["message"]["content"]
data = resp.json()
return data.get("text") or data.get("response") or ""
_CHAT_DISPATCH = {
Provider.OPENAI: _chat_openai,
Provider.GEMINI: _chat_gemini,
Provider.CLAUDE: _chat_claude,
Provider.KIMI: _chat_kimi,
# Dispatch by protocol string, not by enum member — adding a new OpenAI-compatible
# provider to registry.yaml needs no code change here.
_CHAT_BY_PROTOCOL = {
"openai": _chat_openai,
"anthropic": _chat_anthropic,
"gemini": _chat_gemini,
"bedrock-converse": _chat_bedrock,
"portal": _chat_portal,
}
@ -207,9 +333,9 @@ def _sanitize_error(error: str) -> str:
error = re.sub(r'Bearer [^\s\'\"]+', 'Bearer ***REDACTED***', error)
return error
# Default preference order for cascading fallback
# Kimi is primary; Gemini escalates when Kimi quality/availability falls short
_CHAT_PREFERENCE = [Provider.KIMI, Provider.GEMINI, Provider.CLAUDE, Provider.OPENAI]
# Default preference order for cascading fallback is now loaded from
# mcp.providers.json (`_CHAT_PREFERENCE`, set at import time above). Kept here
# only as a comment for readers — see registry.yaml `mcp_chat_fallback`.
async def synthesize(
@ -233,7 +359,7 @@ async def synthesize(
# Build ordered candidate list
if provider:
candidates = [provider]
candidates = [resolve_provider(provider)]
else:
candidates = [p for p in _CHAT_PREFERENCE if p in available_chat_providers()]
@ -243,9 +369,12 @@ async def synthesize(
last_error = None
for candidate in candidates:
cfg = REGISTRY[candidate]
fn = _CHAT_DISPATCH.get(candidate)
cfg = REGISTRY.get(candidate)
if not cfg:
continue
fn = _CHAT_BY_PROTOCOL.get(cfg.protocol)
if not fn:
logger.warning("model_router.no_dispatch", provider=candidate.value, protocol=cfg.protocol)
continue
try:
logger.info("model_router.chat", provider=candidate.value, model=cfg.chat_model)
@ -292,14 +421,13 @@ async def _embed_gemini(cfg: ProviderConfig, texts: list[str]) -> list[list[floa
return [e["values"] for e in resp.json()["embeddings"]]
_EMBED_DISPATCH = {
Provider.OPENAI: _embed_openai,
Provider.GEMINI: _embed_gemini,
# Claude and Kimi don't expose embedding endpoints — they route through LOCAL or OpenAI/Gemini.
# Embedding dispatch is protocol-driven too. OpenAI-compatible embedding
# endpoints (OpenAI, Qwen/DashScope, Alibaba Cloud) share `_embed_openai`.
_EMBED_BY_PROTOCOL = {
"openai": _embed_openai,
"gemini": _embed_gemini,
}
_EMBED_PREFERENCE = [Provider.GEMINI, Provider.OPENAI] # Kimi has no embed endpoint, Gemini leads here
async def embed_texts(
texts: list[str],
@ -315,15 +443,17 @@ async def embed_texts(
Dict with keys: provider, model, embeddings (list[list[float]]), error.
"""
if provider:
candidates = [provider]
candidates = [resolve_provider(provider)]
else:
candidates = [p for p in _EMBED_PREFERENCE
if p in available_embedding_providers() and p != Provider.LOCAL]
last_error = None
for candidate in candidates:
cfg = REGISTRY[candidate]
fn = _EMBED_DISPATCH.get(candidate)
cfg = REGISTRY.get(candidate)
if not cfg:
continue
fn = _EMBED_BY_PROTOCOL.get(cfg.protocol)
if not fn:
continue
try:
@ -369,5 +499,12 @@ def provider_status() -> dict:
"chat_model": cfg.chat_model or None,
"embedding_model": cfg.embedding_model or None,
"max_context_tokens": cfg.max_context,
"protocol": cfg.protocol,
"auth": cfg.auth,
}
status["_meta"] = {
"registry_source": str(_MIRROR) if _MIRROR else "builtin",
"chat_fallback": [p.value for p in _CHAT_PREFERENCE],
"embed_fallback": [p.value for p in _EMBED_PREFERENCE],
}
return status