Robert Helewka
4809d781ff
fast-agent auto-activates its shell execute tool on any agent with skills configured, and Context.no_shell — the only opt-out — has no constructor or config knob. Add install_no_shell(), a wrapper on fast_agent.context.initialize_context that stamps no_shell=True on every context, installed from server.main() when the deployment's agents.yaml sets a truthy top-level no_shell: key. Skill loading via read_skill is unaffected. Default behaviour unchanged for deployments without the key. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
417 lines
16 KiB
Python
417 lines
16 KiB
Python
"""
|
|
Pallas — FastAgent MCP Bridge
|
|
|
|
Reads agent topology from agents.yaml in the working directory and exposes
|
|
each agent as a StreamableHTTP MCP endpoint with an optional registry server.
|
|
|
|
Usage:
|
|
pallas # all agents + registry
|
|
pallas --agent jarvis # single agent mode
|
|
"""
|
|
|
|
import argparse
|
|
import asyncio
|
|
import importlib
|
|
import logging
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
from pallas import _fastagent_patch
|
|
from pallas.log import set_agent_component, set_project, setup_logging
|
|
from pallas.multimodal_server import MultimodalAgentMCPServer
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _config_root() -> Path:
|
|
"""Return the working directory where agents.yaml and fastagent configs live."""
|
|
return Path.cwd()
|
|
|
|
|
|
# ── Configuration ─────────────────────────────────────────────────────────────
|
|
|
|
def _load_deployment_config() -> dict:
|
|
"""Load agents.yaml — single source of truth for deployment topology."""
|
|
config_path = _config_root() / os.environ.get("PALLAS_AGENTS_CONFIG", "agents.yaml")
|
|
if not config_path.exists():
|
|
raise SystemExit(f"deployment config not found: {config_path}")
|
|
|
|
with open(config_path) as f:
|
|
config = yaml.safe_load(f) or {}
|
|
|
|
if "agents" not in config:
|
|
raise SystemExit(f"no 'agents' section in {config_path}")
|
|
|
|
return config
|
|
|
|
|
|
def _build_agents_table(config: dict) -> dict[str, dict]:
|
|
"""Build {name: {module, port, model?, model_capabilities?}} from agents.yaml.
|
|
|
|
The ``model`` and ``model_capabilities`` fields are optional. When ``model``
|
|
is set, Pallas overrides the agent's ``AgentConfig.model`` at startup so
|
|
fast-agent routes that agent to the specified model/provider. When
|
|
``model_capabilities`` is set, those capabilities are used to register the
|
|
model with fast-agent's ``ModelDatabase`` instead of the top-level defaults
|
|
from ``fastagent.config.yaml``.
|
|
"""
|
|
return {
|
|
name: {
|
|
"module": agent["module"],
|
|
"port": agent["port"],
|
|
"model": agent.get("model"),
|
|
"model_capabilities": agent.get("model_capabilities"),
|
|
"max_iterations": agent.get("max_iterations"),
|
|
"streaming_timeout": agent.get("streaming_timeout"),
|
|
"turn_timeout": agent.get("turn_timeout"),
|
|
"loop_repeat_threshold": agent.get("loop_repeat_threshold"),
|
|
}
|
|
for name, agent in config["agents"].items()
|
|
}
|
|
|
|
|
|
def _build_agent_deps(config: dict) -> dict[str, list[str]]:
|
|
"""Build dependency graph from agents.yaml depends_on fields."""
|
|
agents = config["agents"]
|
|
deps: dict[str, list[str]] = {}
|
|
for name, agent in agents.items():
|
|
dep_list = agent.get("depends_on", [])
|
|
if dep_list:
|
|
deps[name] = list(dep_list)
|
|
return deps
|
|
|
|
|
|
# ── Downstream MCP server helpers ─────────────────────────────────────────────
|
|
|
|
def _resolve_downstream_servers(fast_instance) -> dict[str, dict]:
|
|
"""Collect downstream MCP server configs referenced by any agent."""
|
|
server_names = set()
|
|
for agent_data in fast_instance.agents.values():
|
|
config = agent_data.get("config")
|
|
if config:
|
|
for s in getattr(config, "servers", []):
|
|
server_names.add(s)
|
|
|
|
mcp_servers = fast_instance.config.get("mcp", {}).get("servers", {})
|
|
servers = {}
|
|
for name in sorted(server_names):
|
|
cfg = mcp_servers.get(name, {})
|
|
if isinstance(cfg, dict):
|
|
url = cfg.get("url")
|
|
headers = cfg.get("headers") or {}
|
|
else:
|
|
url = getattr(cfg, "url", None)
|
|
headers = getattr(cfg, "headers", None) or {}
|
|
if url:
|
|
servers[name] = {"url": str(url), "headers": dict(headers)}
|
|
return servers
|
|
|
|
|
|
def _preflight_mcp_servers(agent_name: str, servers: dict[str, dict]) -> None:
|
|
"""Warn at startup if MCP server auth headers have unresolved env vars."""
|
|
import re
|
|
|
|
for server_name, cfg in servers.items():
|
|
for header_key, header_val in cfg.get("headers", {}).items():
|
|
unresolved = re.findall(r"\$\{([^}]+)\}", str(header_val))
|
|
for var in unresolved:
|
|
val = os.environ.get(var, "")
|
|
if not val:
|
|
logger.warning(
|
|
"%s → %s: %s references ${%s} but it is not set",
|
|
agent_name, server_name, header_key, var,
|
|
)
|
|
|
|
|
|
# ── Model registration ────────────────────────────────────────────────────────
|
|
|
|
def _register_one_model(model_spec: str, capabilities: dict) -> None:
|
|
"""Register a single unknown model with fast-agent's ModelDatabase.
|
|
|
|
If fast-agent already has a built-in entry for this model we leave it
|
|
alone. Otherwise we register a minimal ``ModelParameters`` using the
|
|
declared capabilities so the model resolves cleanly at agent startup.
|
|
"""
|
|
from fast_agent.llm.model_database import ModelDatabase, ModelParameters
|
|
|
|
model_name = model_spec.split(".", 1)[-1] if "." in model_spec else model_spec
|
|
|
|
if ModelDatabase.get_model_params(model_name) is not None:
|
|
return
|
|
|
|
is_vision = capabilities.get("vision", False)
|
|
context_window = capabilities.get("context_window", 131072)
|
|
max_output_tokens = capabilities.get("max_output_tokens", 16384)
|
|
|
|
if is_vision:
|
|
tokenizes = list(ModelDatabase.QWEN_MULTIMODAL)
|
|
logger.info("Registered model '%s' with vision capabilities", model_name)
|
|
else:
|
|
tokenizes = list(ModelDatabase.TEXT_ONLY)
|
|
logger.info("Registered model '%s' as text-only", model_name)
|
|
|
|
ModelDatabase.register_runtime_model_params(
|
|
model_name,
|
|
ModelParameters(
|
|
context_window=context_window,
|
|
max_output_tokens=max_output_tokens,
|
|
tokenizes=tokenizes,
|
|
),
|
|
)
|
|
|
|
|
|
|
|
def _register_unknown_models(deployment_config: dict) -> None:
|
|
"""Register runtime model params for models not in fast-agent's ModelDatabase.
|
|
|
|
Registers the ``default_model`` from ``fastagent.config.yaml`` plus every
|
|
per-agent ``model`` declared in ``agents.yaml``. Capabilities are resolved
|
|
per model: if the agent carries its own ``model_capabilities`` block, those
|
|
take effect; otherwise the top-level ``model_capabilities`` from
|
|
``fastagent.config.yaml`` apply.
|
|
|
|
Also auto-detects an AWS Bedrock Mantle ``anthropic.base_url`` and installs
|
|
the Mantle compatibility shims (wire-name prefix and ``tool_use.caller``
|
|
strip) via :mod:`pallas.mantle_shims`. No config flag needed — Pallas
|
|
reads the base_url and does the right thing.
|
|
"""
|
|
from pallas import mantle_shims
|
|
|
|
fastagent_config_path = _config_root() / "fastagent.config.yaml"
|
|
if not fastagent_config_path.exists():
|
|
return
|
|
|
|
with open(fastagent_config_path) as f:
|
|
fa_config = yaml.safe_load(f) or {}
|
|
|
|
anthropic_base_url = fa_config.get("anthropic", {}).get("base_url", "")
|
|
if mantle_shims.maybe_install(anthropic_base_url):
|
|
logger.info(
|
|
"Detected Bedrock Mantle endpoint (%s); installed fast-agent shims.",
|
|
anthropic_base_url,
|
|
)
|
|
|
|
default_model = fa_config.get("default_model", "")
|
|
default_capabilities = fa_config.get("model_capabilities", {})
|
|
|
|
seen: set[str] = set()
|
|
|
|
if default_model:
|
|
_register_one_model(default_model, default_capabilities)
|
|
seen.add(default_model)
|
|
|
|
for agent_name, agent in deployment_config.get("agents", {}).items():
|
|
agent_model = agent.get("model")
|
|
if not agent_model or agent_model in seen:
|
|
continue
|
|
agent_caps = agent.get("model_capabilities") or default_capabilities
|
|
_register_one_model(agent_model, agent_caps)
|
|
seen.add(agent_model)
|
|
|
|
|
|
# ── Agent lifecycle ───────────────────────────────────────────────────────────
|
|
|
|
async def _preflight(deployment_config: dict) -> None:
|
|
from pallas.health import validate_llm_providers
|
|
|
|
_register_unknown_models(deployment_config)
|
|
await validate_llm_providers()
|
|
|
|
|
|
async def _start_agent(name: str, agents: dict[str, dict]) -> None:
|
|
from pallas.health import register_health_tool
|
|
|
|
# Tag every log record emitted from this asyncio task with the agent
|
|
# name (``component`` field). ContextVar scope is per-task, so each
|
|
# agent started from ``asyncio.gather`` carries its own value and the
|
|
# registry's own task sees ``component="runtime"`` by default. Fast-
|
|
# agent, fastmcp, anthropic and openai records all inherit this tag
|
|
# because the _StaticFieldsFilter is attached to every handler.
|
|
set_agent_component(name)
|
|
|
|
entry = agents[name]
|
|
module_path = entry["module"]
|
|
port = entry["port"]
|
|
model_override = entry.get("model")
|
|
|
|
module = importlib.import_module(module_path)
|
|
fast_instance = module.fast
|
|
|
|
if model_override:
|
|
for agent_data in fast_instance.agents.values():
|
|
agent_cfg = agent_data.get("config")
|
|
if agent_cfg is not None:
|
|
agent_cfg.model = model_override
|
|
logger.info("%s model override → %s", name, model_override)
|
|
|
|
logger.info("Starting %s agent on port %d", name, port)
|
|
|
|
async with fast_instance.run():
|
|
primary_instance = fast_instance._server_managed_instances[0]
|
|
|
|
# Stateless per request: each MCP `tools/call` gets a freshly-created
|
|
# agent instance which is disposed immediately after the response.
|
|
# Conversation history is owned by the caller (Daedalus) and supplied
|
|
# on every turn via the `history` argument on `send_message` — see
|
|
# multimodal_server.MultimodalAgentMCPServer.register_agent_tools.
|
|
#
|
|
# Why this matters:
|
|
# * "shared" leaks one conversation's history into the next
|
|
# because all callers see the same `agent.message_history`.
|
|
# * "shared" also silently loses everything on process restart,
|
|
# breaking the "Pallas is ephemeral" contract.
|
|
# With "request" the Pallas process holds no per-conversation state
|
|
# and the LLM sees exactly what Daedalus asks it to see.
|
|
request_limits = {
|
|
k: entry[k]
|
|
for k in (
|
|
"max_iterations",
|
|
"streaming_timeout",
|
|
"turn_timeout",
|
|
"loop_repeat_threshold",
|
|
)
|
|
if entry.get(k) is not None
|
|
}
|
|
|
|
server = MultimodalAgentMCPServer(
|
|
primary_instance=primary_instance,
|
|
create_instance=fast_instance._server_instance_factory,
|
|
dispose_instance=fast_instance._server_instance_dispose,
|
|
instance_scope="request",
|
|
server_name=f"{fast_instance.name}-MCP-Server",
|
|
host="0.0.0.0",
|
|
get_registry_version=fast_instance._get_registry_version,
|
|
request_limits=request_limits,
|
|
)
|
|
|
|
downstream_servers = _resolve_downstream_servers(fast_instance)
|
|
_preflight_mcp_servers(name, downstream_servers)
|
|
register_health_tool(server.mcp_server, downstream_servers, agent_name=name)
|
|
|
|
await server.run_async(transport="http", host="0.0.0.0", port=port)
|
|
|
|
|
|
async def _wait_for_agent(
|
|
name: str,
|
|
agents: dict[str, dict],
|
|
timeout: float = 60.0,
|
|
) -> None:
|
|
import httpx
|
|
|
|
port = agents[name]["port"]
|
|
url = f"http://127.0.0.1:{port}/ready"
|
|
deadline = asyncio.get_event_loop().time() + timeout
|
|
while asyncio.get_event_loop().time() < deadline:
|
|
try:
|
|
async with httpx.AsyncClient(timeout=2.0) as client:
|
|
await client.get(url)
|
|
logger.info("%s is ready", name)
|
|
return
|
|
except Exception:
|
|
await asyncio.sleep(1.0)
|
|
logger.warning("%s did not become ready within %.0fs", name, timeout)
|
|
|
|
|
|
async def _run_single(name: str, agents: dict[str, dict], deployment_config: dict) -> None:
|
|
await _preflight(deployment_config)
|
|
await _start_agent(name, agents)
|
|
|
|
|
|
async def _start_all(config: dict) -> None:
|
|
from pallas.registry import run_registry
|
|
|
|
agents = _build_agents_table(config)
|
|
agent_deps = _build_agent_deps(config)
|
|
registry_port = config.get("registry_port", 24200)
|
|
|
|
await _preflight(config)
|
|
|
|
# Identify subagents that must start first.
|
|
subagents: set[str] = set()
|
|
for dep_name, dep_list in agent_deps.items():
|
|
if dep_name in agents:
|
|
for sub in dep_list:
|
|
if sub in agents:
|
|
subagents.add(sub)
|
|
|
|
long_running: list[asyncio.Task] = []
|
|
|
|
long_running.append(
|
|
asyncio.create_task(run_registry(port=registry_port))
|
|
)
|
|
|
|
for name in sorted(subagents):
|
|
long_running.append(asyncio.create_task(_start_agent(name, agents)))
|
|
|
|
await asyncio.gather(*[_wait_for_agent(n, agents) for n in subagents])
|
|
|
|
for name in agents:
|
|
if name not in subagents:
|
|
long_running.append(asyncio.create_task(_start_agent(name, agents)))
|
|
|
|
await asyncio.gather(*long_running)
|
|
|
|
|
|
# ── CLI ───────────────────────────────────────────────────────────────────────
|
|
|
|
def main() -> None:
|
|
config = _load_deployment_config()
|
|
agents = _build_agents_table(config)
|
|
registry_port = config.get("registry_port", 24200)
|
|
deploy_name = config.get("name", "pallas")
|
|
|
|
parser = argparse.ArgumentParser(
|
|
description=f"{deploy_name.title()} — FastAgent MCP Bridge",
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
epilog=(
|
|
"Port assignments:\n"
|
|
+ "\n".join(
|
|
f" {name:16s} port {entry['port']}"
|
|
for name, entry in agents.items()
|
|
)
|
|
+ f"\n {'registry':16s} port {registry_port}"
|
|
),
|
|
)
|
|
parser.add_argument(
|
|
"--agent",
|
|
choices=list(agents.keys()),
|
|
metavar="AGENT",
|
|
help="Start a specific agent (default: all). Choices: %(choices)s",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
# Project label (e.g. "kottos", "mentor", "iolaus") is read from
|
|
# ``agents.yaml`` and stamped onto every log record via
|
|
# ``pallas.log._StaticFieldsFilter``. Must be set before
|
|
# ``setup_logging()`` so the "log file: …" bootstrap record already
|
|
# carries the right project label.
|
|
set_project(deploy_name)
|
|
|
|
setup_logging()
|
|
|
|
# Team-level policy: a truthy top-level ``no_shell:`` in agents.yaml
|
|
# keeps fast-agent's shell tool off every agent in this deployment
|
|
# (it otherwise auto-activates on agents with skills configured).
|
|
# After setup_logging() so the install log record isn't lost.
|
|
if config.get("no_shell"):
|
|
_fastagent_patch.install_no_shell()
|
|
|
|
if args.agent:
|
|
port = agents[args.agent]["port"]
|
|
logger.info("Starting %s agent on port %d", args.agent, port)
|
|
asyncio.run(_run_single(args.agent, agents, config))
|
|
else:
|
|
logger.info("Starting all agents + registry for %s", deploy_name)
|
|
logger.info(
|
|
"registry → http://0.0.0.0:%d/.well-known/mcp/server.json", registry_port
|
|
)
|
|
for name, entry in agents.items():
|
|
logger.info("%-16s → http://0.0.0.0:%d/mcp", name, entry["port"])
|
|
asyncio.run(_start_all(config))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|