tlamatini_mcp_server.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
# ═══════════════════════════════════════════════════════════════════
# ✦ T L A M A T I N I ✦ — "one who knows"
#
# Created by Angela López Mendoza · @angelahack1
# Developer · Architect · Creator of Tlamatini
#
# Every line of this file was written by Angela López Mendoza.
# ═══════════════════════════════════════════════════════════════════
# Tlamatini Author Banner — do not remove (releases scrub the name automatically)
"""Tlamatini Agents — MCP server.
Exposes EVERY Tlamatini pool agent (Executer, Pythonxer, Croner, ACPXer,
STM32er, ESP32er, Arduiner, Shoter, Playwrighter, Kalier, Blenderer,
Talker, Whisperer, Discoverer, MCP-Doctor, NetSpeed-Calculator, and every
other complete live agent directory) as an MCP tool so an
MCP client (Claude Code, etc.) can drive them directly.
Agent discovery is fully DYNAMIC: every directory under
``Tlamatini/agent/agents/`` that holds both ``<name>.py`` and ``config.yaml``
is surfaced as a tool automatically at startup — adding a new pool agent
requires NO edit here, it appears on the next server launch.
It does NOT import the Django app. For each call it performs the exact
"launcher dance" Tlamatini's pool uses:
1. copy agent/agents/<name>/ -> Temp/mcp_agent_runs/<name>__<runid>/
2. write the tool args into that copy's config.yaml (deep-merged onto the
template defaults; empty values are dropped so template defaults survive)
3. run python <name>.py (in the copied dir)
4. read <name>__<runid>.log (the agent writes its result there)
Short agents are awaited (default) and their full log is returned. Known
long-running agents (Croner, FlowHypervisor, TeleTlamatini, monitors, …) and
any call with wait=false return a run_id you poll with tlamatini_run_log /
tlamatini_run_status / tlamatini_run_stop.
Self-contained: only needs `mcp` + `pyyaml` (psutil used for tree-kill if
present). Resolves the agents directory relative to THIS file, so the working
directory of the launching client is irrelevant.
"""
from __future__ import annotations
import asyncio
import json
import os
import shutil
import subprocess
import sys
import time
from typing import Any, Dict, List, Optional
import yaml
import mcp.types as types
from mcp.server import Server
from mcp.server.stdio import stdio_server
# --------------------------------------------------------------------------- #
# Paths (resolved relative to this file — client cwd does not matter)
# --------------------------------------------------------------------------- #
HERE = os.path.dirname(os.path.abspath(__file__))
AGENTS_ROOT = os.path.join(HERE, "Tlamatini", "agent", "agents")
SKILLS_ROOT = os.path.join(HERE, "Tlamatini", "agent", "skills_pkg")
# Source-mode app-Temp is the repo-root /Temp (gitignored); see the 2026-06-02
# Temp policy in CLAUDE.md. Runtime agent copies live under it so they never
# pollute the working tree.
TEMP_ROOT = os.path.join(HERE, "Temp")
RUNS_ROOT = os.path.join(TEMP_ROOT, "mcp_agent_runs")
PYTHON_EXE = sys.executable
# Connection-wiring fields: never surfaced as tool params (single-agent runs
# don't wire flows). Left at their template defaults unless explicitly overridden.
CONNECTION_FIELDS = {
"source_agents", "target_agents", "output_agents",
"source_agent_1", "source_agent_2",
"target_agents_a", "target_agents_b", "target_agents_l", "target_agents_g",
}
# Agents that never finish on their own → default to background (wait=false).
LONG_RUNNING = {
"croner", "flowhypervisor", "teletlamatini",
"recmailer",
"monitor_log", "monitor_netstat",
}
DEFAULT_TIMEOUT = 180 # seconds to await a short agent before backgrounding it
# In-memory registry of launched runs (the stdio server is one long-lived process)
_RUNS: Dict[str, Dict[str, Any]] = {}
_RUN_COUNTER = 0
# --------------------------------------------------------------------------- #
# Discovery
# --------------------------------------------------------------------------- #
def discover_agents() -> Dict[str, Dict[str, Any]]:
"""Return {agent_name: {dir, config(dict), params(list)}} for every runnable
agent (a directory holding both <name>.py and config.yaml)."""
agents: Dict[str, Dict[str, Any]] = {}
if not os.path.isdir(AGENTS_ROOT):
return agents
for name in sorted(os.listdir(AGENTS_ROOT)):
d = os.path.join(AGENTS_ROOT, name)
script = os.path.join(d, f"{name}.py")
cfg = os.path.join(d, "config.yaml")
if not (os.path.isfile(script) and os.path.isfile(cfg)):
continue
try:
with open(cfg, "r", encoding="utf-8") as f:
config = yaml.safe_load(f) or {}
except Exception:
config = {}
if not isinstance(config, dict):
config = {}
params = [k for k in config.keys() if k not in CONNECTION_FIELDS]
agents[name] = {"dir": d, "config": config, "params": params}
return agents
def discover_skills() -> Dict[str, Dict[str, Any]]:
"""Return {skill_name: {dir, path, description, runtime, frontmatter}} for
every SKILL.md package under skills_pkg/. Skills are PROCEDURAL runbooks the
MCP client follows — there is no in-process harness here, so the capability
surfaced is: enumerate them (tlamatini_list_skills) and fetch a skill's full
SKILL.md to follow step-by-step (tlamatini_read_skill)."""
skills: Dict[str, Dict[str, Any]] = {}
if not os.path.isdir(SKILLS_ROOT):
return skills
for name in sorted(os.listdir(SKILLS_ROOT)):
d = os.path.join(SKILLS_ROOT, name)
sk = os.path.join(d, "SKILL.md")
if not os.path.isfile(sk):
continue
fm: Dict[str, Any] = {}
try:
with open(sk, "r", encoding="utf-8") as f:
text = f.read()
if text.startswith("---"):
end = text.find("\n---", 3)
if end != -1:
loaded = yaml.safe_load(text[3:end])
fm = loaded if isinstance(loaded, dict) else {}
except Exception:
fm = {}
desc = str(fm.get("description", "")) if fm else ""
try:
runtime = str(((fm.get("metadata") or {}).get("tlamatini") or {}).get("runtime", ""))
except Exception:
runtime = ""
sk_name = str(fm.get("name") or name)
skills[sk_name] = {"dir": d, "path": sk, "description": desc,
"runtime": runtime, "frontmatter": fm}
return skills
def _read_skill_md(info: Dict[str, Any], max_chars: int = 40000) -> str:
try:
with open(info["path"], "r", encoding="utf-8", errors="replace") as f:
data = f.read()
except Exception as e:
return f"(could not read SKILL.md: {e})"
if len(data) > max_chars:
return data[:max_chars] + "\n…(truncated)…"
return data
def _json_type(value: Any) -> str:
if isinstance(value, bool):
return "boolean"
if isinstance(value, int):
return "integer"
if isinstance(value, float):
return "number"
if isinstance(value, list):
return "array"
if isinstance(value, dict):
return "object"
return "string"
def build_tool(name: str, info: Dict[str, Any]) -> types.Tool:
config = info["config"]
props: Dict[str, Any] = {}
for key in info["params"]:
default = config.get(key)
t = _json_type(default)
schema: Dict[str, Any] = {"type": t}
if t in ("array", "object"):
# keep loose so the client can pass arbitrary structure
schema = {"type": t}
schema["description"] = f"{key} (template default: {json.dumps(default, default=str)})"
props[key] = schema
if name == 'video_analyzer':
props['analysis_type'] = {
'type': 'string', 'enum': ['robotics', 'transcription', 'summary'],
'description': 'robotics (default): physical-motion verdict; transcription: speech from video audio tracks; '
'summary: timestamped visual evidence plus speech and detailed synthesis.',
}
props["config"] = {
"type": "object",
"description": "Free-form overrides deep-merged onto config.yaml "
"(use for nested/uncommon keys, or to set source_agents/"
"target_agents wiring).",
}
props["wait"] = {
"type": "boolean",
"description": f"Wait for the agent to finish and return its full log. "
f"Default {name not in LONG_RUNNING} for this agent.",
}
props["timeout_seconds"] = {
"type": "integer",
"description": f"Max seconds to wait before backgrounding (default {DEFAULT_TIMEOUT}).",
}
param_list = ", ".join(info["params"]) or "(no configurable parameters)"
lr = " [LONG-RUNNING: defaults to background — poll with tlamatini_run_log]" if name in LONG_RUNNING else ""
desc = (f"Run the Tlamatini **{name}** pool agent.{lr}\n"
f"Configurable parameters: {param_list}.\n"
f"Returns the agent's execution log. Backgrounds (returns a run_id) "
f"on timeout or wait=false.")
if name == 'video_analyzer':
desc += ('\nChoose analysis_type explicitly for transcription or summary; both accept static scenes and '
'bypass the robotics gate. audio_tracks selects all tracks or audio indexes such as 0,1. '
'Local faster-whisper uses GPU auto/CPU fallback; no microphone is opened. Content results '
'include transcript, summary, timestamps, audio_status, artifact paths and TLM_ANALYSIS routing. '
'Inspect partial/error outcomes and coverage limits; only robotics uses TLM_VERDICT.')
return types.Tool(
name=name,
description=desc,
inputSchema={"type": "object", "properties": props, "additionalProperties": True},
)
# --------------------------------------------------------------------------- #
# Launcher
# --------------------------------------------------------------------------- #
def _deep_merge(base: Dict[str, Any], overrides: Dict[str, Any]) -> Dict[str, Any]:
for k, v in overrides.items():
if v is None:
continue # never let an empty value clobber a template default
if isinstance(v, str) and v == "":
continue
if isinstance(v, dict) and isinstance(base.get(k), dict):
_deep_merge(base[k], v)
else:
base[k] = v
return base
def _ignore(_dir: str, names: List[str]) -> List[str]:
drop = []
for n in names:
if n == "__pycache__" or n.endswith(".log") or n.endswith(".pid"):
drop.append(n)
elif n.startswith("reanim") or n == "agent.pid":
drop.append(n)
return drop
def _log_path(run_dir: str) -> str:
return os.path.join(run_dir, os.path.basename(run_dir) + ".log")
def _read_log(run_dir: str, max_chars: int = 12000) -> str:
p = _log_path(run_dir)
try:
with open(p, "r", encoding="utf-8", errors="replace") as f:
data = f.read()
except FileNotFoundError:
return "(no log produced yet)"
except Exception as e:
return f"(could not read log: {e})"
if len(data) > max_chars:
return "…(truncated)…\n" + data[-max_chars:]
return data
def run_agent_blocking(agent: str, info: Dict[str, Any], overrides: Dict[str, Any],
wait: bool, timeout: int) -> Dict[str, Any]:
global _RUN_COUNTER
os.makedirs(RUNS_ROOT, exist_ok=True)
_RUN_COUNTER += 1
run_id = f"{agent}__{int(time.time())}_{_RUN_COUNTER}"
run_dir = os.path.join(RUNS_ROOT, run_id)
shutil.copytree(info["dir"], run_dir, ignore=_ignore)
# These runs live outside the agents tree, so carry the portable registry
# and resolve the app defaults before applying explicit per-call overrides.
import importlib.util
registry_path = os.path.join(AGENTS_ROOT, 'model_settings.py')
registry_spec = importlib.util.spec_from_file_location('mcp_model_settings', registry_path)
registry = importlib.util.module_from_spec(registry_spec)
registry_spec.loader.exec_module(registry)
if agent in registry.AGENTS:
shutil.copy2(registry_path, os.path.join(run_dir, 'model_settings.py'))
# merge overrides onto the template config.yaml
cfg_path = os.path.join(run_dir, "config.yaml")
try:
with open(cfg_path, "r", encoding="utf-8") as f:
cfg = yaml.safe_load(f) or {}
except Exception:
cfg = dict(info["config"])
cfg = registry.resolve_agent_models(agent, cfg, agent_file=registry_path)
_deep_merge(cfg, overrides or {})
with open(cfg_path, "w", encoding="utf-8") as f:
yaml.dump(cfg, f, default_flow_style=False, allow_unicode=True, sort_keys=False)
env = os.environ.copy()
env["TLAMATINI_TEMP"] = TEMP_ROOT
env["PYTHONUTF8"] = "1"
env["PYTHONIOENCODING"] = "utf-8"
env.setdefault('CONFIG_PATH', os.path.join(os.path.dirname(AGENTS_ROOT), 'config.json'))
env.pop("AGENT_REANIMATED", None)
proc = subprocess.Popen(
[PYTHON_EXE, f"{agent}.py"],
cwd=run_dir,
env=env,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
stdin=subprocess.DEVNULL,
)
_RUNS[run_id] = {"proc": proc, "dir": run_dir, "agent": agent, "started": time.time()}
echoed = _redact_bulky(cfg)
if not wait:
return {"status": "launched", "run_id": run_id, "agent": agent,
"note": "Running in background. Use tlamatini_run_log / "
"tlamatini_run_status / tlamatini_run_stop.",
"config_used": echoed, "run_dir": run_dir}
try:
rc = proc.wait(timeout=timeout)
return {"status": "finished", "run_id": run_id, "agent": agent,
"return_code": rc, "config_used": echoed, "log": _read_log(run_dir)}
except subprocess.TimeoutExpired:
return {"status": "running", "run_id": run_id, "agent": agent,
"note": f"Still running after {timeout}s — left in background. "
f"Poll with tlamatini_run_log('{run_id}').",
"config_used": echoed, "log_excerpt": _read_log(run_dir, 4000)}
_ECHO_VALUE_LIMIT = 600
def _redact_bulky(value, _depth: int = 0):
"""Shrink huge values before they are echoed back as ``config_used``.
The MCP result echoes the resolved config so the caller can see exactly
what the agent ran with. That is genuinely useful for a 20-character
``pattern`` -- and actively harmful for a 46 KB ``content`` (File-Creator)
or a 10 KB ``input_text`` (LaTeXer): the echo alone can blow the caller's
response budget and lose the run's real log. Long strings are therefore
replaced by a short, HONEST placeholder that still states the true size,
so nothing is silently misrepresented.
"""
if _depth > 6:
return "<nested too deeply>"
if isinstance(value, str):
if len(value) > _ECHO_VALUE_LIMIT:
return "%s... <redacted from echo: %d chars total>" % (
value[:_ECHO_VALUE_LIMIT], len(value))
return value
if isinstance(value, dict):
return {k: _redact_bulky(v, _depth + 1) for k, v in value.items()}
if isinstance(value, list):
if len(value) > 200:
head = [_redact_bulky(v, _depth + 1) for v in value[:200]]
head.append("<redacted from echo: %d items total>" % len(value))
return head
return [_redact_bulky(v, _depth + 1) for v in value]
return value
def _kill_tree(proc: subprocess.Popen) -> None:
try:
import psutil
try:
parent = psutil.Process(proc.pid)
except psutil.NoSuchProcess:
return
for child in parent.children(recursive=True):
try:
child.terminate()
except Exception:
pass
parent.terminate()
gone, alive = psutil.wait_procs([parent], timeout=3)
for p in alive:
try:
p.kill()
except Exception:
pass
except Exception:
try:
proc.terminate()
except Exception:
pass
# --------------------------------------------------------------------------- #
# MCP server
# --------------------------------------------------------------------------- #
server = Server("tlamatini")
_AGENTS = discover_agents()
_SKILLS = discover_skills()
# Self-contained ACPX runtime (drives external coding-agent CLIs as child
# processes). Optional import so a load failure degrades the acp_* tools to a
# clear error instead of killing the whole server.
try:
from tlamatini_acpx import AcpxManager
_ACPX: Optional[Any] = AcpxManager(os.path.join(TEMP_ROOT, "acpx_sessions"))
_ACPX_IMPORT_ERROR = ""
except Exception as _acpx_err: # noqa: BLE001
_ACPX = None
_ACPX_IMPORT_ERROR = repr(_acpx_err)
_ACPX_TOOL_NAMES = {
"acp_doctor", "list_acp_agents", "acp_spawn", "acp_send", "acp_send_and_wait",
"acp_relay", "acp_kill", "acp_transcript", "acp_session_status", "acp_list_sessions",
}
MANAGEMENT_TOOLS = {
"tlamatini_list_agents", "tlamatini_run_status",
"tlamatini_run_log", "tlamatini_run_stop", "tlamatini_list_runs",
"tlamatini_list_skills", "tlamatini_read_skill",
} | _ACPX_TOOL_NAMES
@server.list_tools()
async def list_tools() -> List[types.Tool]:
tools = [build_tool(name, info) for name, info in _AGENTS.items()]
tools += [
types.Tool(
name="tlamatini_list_agents",
description="List every available Tlamatini agent tool and its configurable parameters.",
inputSchema={"type": "object", "properties": {}},
),
types.Tool(
name="tlamatini_run_status",
description="Status (alive / finished + return code) of a background agent run.",
inputSchema={"type": "object",
"properties": {"run_id": {"type": "string"}},
"required": ["run_id"]},
),
types.Tool(
name="tlamatini_run_log",
description="Read the log of a (running or finished) agent run.",
inputSchema={"type": "object",
"properties": {"run_id": {"type": "string"},
"max_chars": {"type": "integer"}},
"required": ["run_id"]},
),
types.Tool(
name="tlamatini_run_stop",
description="Terminate a background agent run (kills the process tree).",
inputSchema={"type": "object",
"properties": {"run_id": {"type": "string"}},
"required": ["run_id"]},
),
types.Tool(
name="tlamatini_list_runs",
description="List all agent runs launched in this session and whether they are alive.",
inputSchema={"type": "object", "properties": {}},
),
types.Tool(
name="tlamatini_list_skills",
description="List every Tlamatini SKILL.md package (procedural runbook): "
"name, description and runtime. Use tlamatini_read_skill to fetch "
"a skill's full body and follow it step-by-step.",
inputSchema={"type": "object",
"properties": {"filter_keywords": {
"type": "string",
"description": "Optional space-separated terms; only skills whose "
"name+description contain ALL terms are returned."}}},
),
types.Tool(
name="tlamatini_read_skill",
description="Return the full SKILL.md (frontmatter + runbook body) for one skill, "
"so an MCP client can follow the procedure. Skills are runbooks, not "
"executable code — this fetches the instructions to follow.",
inputSchema={"type": "object",
"properties": {"skill_name": {"type": "string"},
"max_chars": {"type": "integer"}},
"required": ["skill_name"]},
),
types.Tool(
name="acp_doctor",
description="ACPX health probe: list external coding-agent CLIs (claude / codex / "
"cursor / gemini / qwen / tlamatini / kiro / kimi / iflow / kilocode / "
"opencode / pi / droid / copilot) and whether each is resolvable on PATH. "
"Call this FIRST in any ACPX flow. WARNING: resolvable only means the "
"binary EXISTS -- a CLI with a dead API key, an invalid config file or no "
"credit still resolves and still answers --version. Pass deep=true to send "
"each oneshot-prompt agent a REAL one-line prompt and get a per-agent "
"readiness block naming the actual failure (AUTH_FAILED / CONFIG_INVALID / "
"NO_CREDIT / USAGE_LIMIT / PERMISSION_BLOCKED / NO_OUTPUT / ...). deep "
"costs real model quota, so use it deliberately; results are cached 10 min.",
inputSchema={"type": "object",
"properties": {"agent_id": {"type": "string"},
"deep": {"type": "boolean"},
"deep_timeout_seconds": {"type": "number"}}},
),
types.Tool(
name="list_acp_agents",
description="Enumerate ACPX agent_ids and their transport (cheaper than acp_doctor; no PATH probe).",
inputSchema={"type": "object", "properties": {}},
),
types.Tool(
name="acp_spawn",
description="Spawn an external coding-agent CLI as an ACPX child and dispatch a task. "
"Returns a session_id (+ captured events / last_assistant_text). "
"agent_id e.g. 'claude','gemini','cursor','codex','qwen'.",
inputSchema={"type": "object",
"properties": {"agent_id": {"type": "string"}, "task": {"type": "string"},
"cwd": {"type": "string"}, "mode": {"type": "string"},
"command": {"type": "string"},
"timeout_seconds": {"type": "number"},
"idle_seconds": {"type": "number"},
"startup_grace_seconds": {"type": "number"}},
"required": ["agent_id", "task"]},
),
types.Tool(
name="acp_send",
description="Send a follow-up turn to an existing ACPX session. (oneshot-prompt agents "
"re-spawn statelessly per turn — carry context in the text.)",
inputSchema={"type": "object",
"properties": {"session_id": {"type": "string"}, "text": {"type": "string"},
"timeout_seconds": {"type": "number"},
"idle_seconds": {"type": "number"},
"startup_grace_seconds": {"type": "number"}},
"required": ["session_id", "text"]},
),
types.Tool(
name="acp_send_and_wait",
description="Send a turn and BLOCK until the child settles (idle rule). "
"Prefer this for 'wait for the full answer' prompts.",
inputSchema={"type": "object",
"properties": {"session_id": {"type": "string"}, "text": {"type": "string"},
"until_idle_seconds": {"type": "number"},
"max_wait_seconds": {"type": "number"}},
"required": ["session_id", "text"]},
),
types.Tool(
name="acp_relay",
description="Hand off one session's last assistant text (or full transcript) to another "
"session and wait for it to settle. transform: last_assistant_text | full_transcript. "
"One call replaces acp_transcript -> string-munge -> acp_send.",
inputSchema={"type": "object",
"properties": {"session_id_src": {"type": "string"},
"session_id_dst": {"type": "string"},
"transform": {"type": "string"},
"prefix": {"type": "string"}, "suffix": {"type": "string"},
"until_idle_seconds": {"type": "number"},
"max_wait_seconds": {"type": "number"}},
"required": ["session_id_src", "session_id_dst"]},
),
types.Tool(
name="acp_transcript",
description="Read an ACPX session's on-disk NDJSON transcript. direction: all | in | out.",
inputSchema={"type": "object",
"properties": {"session_id": {"type": "string"},
"max_chars": {"type": "integer"},
"direction": {"type": "string"}},
"required": ["session_id"]},
),
types.Tool(
name="acp_session_status",
description="Status of one ACPX session (alive / closed / events_total / last assistant text).",
inputSchema={"type": "object", "properties": {"session_id": {"type": "string"}},
"required": ["session_id"]},
),
types.Tool(
name="acp_list_sessions",
description="List every ACPX session spawned in this server process and whether it is alive.",
inputSchema={"type": "object", "properties": {}},
),
types.Tool(
name="acp_kill",
description="Terminate an ACPX session's child-process tree. Always kill sessions you spawned.",
inputSchema={"type": "object", "properties": {"session_id": {"type": "string"}},
"required": ["session_id"]},
),
]
return tools
def _result(obj: Any) -> List[types.TextContent]:
return [types.TextContent(type="text", text=json.dumps(obj, indent=2, default=str))]
@server.call_tool()
async def call_tool(name: str, arguments: Optional[Dict[str, Any]]) -> List[types.TextContent]:
arguments = arguments or {}
if name == "tlamatini_list_agents":
return _result({"agents": [
{"name": n, "parameters": info["params"],
"long_running": n in LONG_RUNNING}
for n, info in _AGENTS.items()
]})
if name == "tlamatini_list_runs":
out = []
for rid, r in _RUNS.items():
rc = r["proc"].poll()
out.append({"run_id": rid, "agent": r["agent"],
"alive": rc is None, "return_code": rc})
return _result({"runs": out})
if name == "tlamatini_run_status":
rid = arguments.get("run_id", "")
r = _RUNS.get(rid)
if not r:
return _result({"error": f"unknown run_id {rid!r}"})
rc = r["proc"].poll()
return _result({"run_id": rid, "agent": r["agent"],
"alive": rc is None, "return_code": rc})
if name == "tlamatini_run_log":
rid = arguments.get("run_id", "")
r = _RUNS.get(rid)
if not r:
return _result({"error": f"unknown run_id {rid!r}"})
return _result({"run_id": rid, "agent": r["agent"],
"alive": r["proc"].poll() is None,
"log": _read_log(r["dir"], int(arguments.get("max_chars", 12000)))})
if name == "tlamatini_run_stop":
rid = arguments.get("run_id", "")
r = _RUNS.get(rid)
if not r:
return _result({"error": f"unknown run_id {rid!r}"})
_kill_tree(r["proc"])
return _result({"run_id": rid, "agent": r["agent"], "stopped": True})
if name == "tlamatini_list_skills":
kw = str(arguments.get("filter_keywords", "")).lower().strip()
terms = kw.split() if kw else []
out = []
for sn, si in _SKILLS.items():
blob = f"{sn} {si['description']}".lower()
if terms and not all(t in blob for t in terms):
continue
out.append({"name": sn, "description": si["description"],
"runtime": si["runtime"]})
return _result({"skills": out, "count": len(out)})
if name == "tlamatini_read_skill":
sn = arguments.get("skill_name", "")
si = _SKILLS.get(sn)
if not si:
return _result({"error": f"unknown skill {sn!r}",
"available": sorted(_SKILLS.keys())})
return _result({"name": sn, "description": si["description"],
"runtime": si["runtime"], "path": si["path"],
"skill_md": _read_skill_md(si, int(arguments.get("max_chars", 40000)))})
if name == "tlamatini_list_skills":
kw = str(arguments.get("filter_keywords", "")).lower().strip()
terms = kw.split() if kw else []
out = []
for sn, si in _SKILLS.items():
blob = f"{sn} {si['description']}".lower()
if terms and not all(t in blob for t in terms):
continue
out.append({"name": sn, "description": si["description"], "runtime": si["runtime"]})
return _result({"skills": out, "count": len(out)})
if name == "tlamatini_read_skill":
sn = arguments.get("skill_name", "")
si = _SKILLS.get(sn)
if not si:
return _result({"error": f"unknown skill {sn!r}", "available": sorted(_SKILLS.keys())})
return _result({"name": sn, "description": si["description"], "runtime": si["runtime"],
"path": si["path"],
"skill_md": _read_skill_md(si, int(arguments.get("max_chars", 40000)))})
if name in _ACPX_TOOL_NAMES:
if _ACPX is None:
return _result({"ok": False, "code": "ACPX_UNAVAILABLE",
"reason": f"ACPX runtime failed to load: {_ACPX_IMPORT_ERROR}"})
try:
if name == "acp_doctor":
return _result(_ACPX.doctor(
arguments.get("agent_id", ""),
deep=bool(arguments.get("deep", False)),
deep_timeout_seconds=float(
arguments.get("deep_timeout_seconds") or 60.0)))
if name == "list_acp_agents":
return _result(_ACPX.list_agents())
if name == "acp_spawn":
return _result(_ACPX.spawn(
arguments.get("agent_id", "claude"), arguments.get("task", ""),
cwd=arguments.get("cwd", ""), mode=arguments.get("mode", "session"),
command=arguments.get("command", ""),
timeout_seconds=arguments.get("timeout_seconds", 0),
idle_seconds=arguments.get("idle_seconds", 0),
startup_grace_seconds=arguments.get("startup_grace_seconds", 0)))
if name == "acp_send":
return _result(_ACPX.send(
arguments.get("session_id", ""), arguments.get("text", ""),
timeout_seconds=arguments.get("timeout_seconds", 0),
idle_seconds=arguments.get("idle_seconds", 0),
startup_grace_seconds=arguments.get("startup_grace_seconds", 0)))
if name == "acp_send_and_wait":
return _result(_ACPX.send_and_wait(
arguments.get("session_id", ""), arguments.get("text", ""),
until_idle_seconds=arguments.get("until_idle_seconds", 10),
max_wait_seconds=arguments.get("max_wait_seconds", 180)))
if name == "acp_relay":
return _result(_ACPX.relay(
arguments.get("session_id_src", ""), arguments.get("session_id_dst", ""),
transform=arguments.get("transform", "last_assistant_text"),
prefix=arguments.get("prefix", ""), suffix=arguments.get("suffix", ""),
until_idle_seconds=arguments.get("until_idle_seconds", 10),
max_wait_seconds=arguments.get("max_wait_seconds", 180)))
if name == "acp_kill":
return _result(_ACPX.kill(arguments.get("session_id", "")))
if name == "acp_transcript":
return _result(_ACPX.transcript(
arguments.get("session_id", ""),
max_chars=int(arguments.get("max_chars", 8000)),
direction=arguments.get("direction", "all")))
if name == "acp_session_status":
return _result(_ACPX.session_status(arguments.get("session_id", "")))
if name == "acp_list_sessions":
return _result(_ACPX.list_sessions())
except Exception as e:
return _result({"ok": False, "code": "ACPX_ERROR", "tool": name, "error": repr(e)})
return _result({"ok": False, "code": "UNKNOWN_ACPX_TOOL", "tool": name})
# ---- an agent tool ----
info = _AGENTS.get(name)
if not info:
return _result({"error": f"unknown tool {name!r}"})
reserved = {"config", "wait", "timeout_seconds"}
overrides: Dict[str, Any] = {k: v for k, v in arguments.items() if k not in reserved}
extra = arguments.get("config") or {}
if isinstance(extra, dict):
_deep_merge(overrides, extra)
wait = bool(arguments.get("wait", name not in LONG_RUNNING))
timeout = int(arguments.get("timeout_seconds", DEFAULT_TIMEOUT))
try:
res = await asyncio.to_thread(run_agent_blocking, name, info, overrides, wait, timeout)
except Exception as e:
return _result({"status": "error", "agent": name, "error": repr(e)})
return _result(res)
async def _amain() -> None:
async with stdio_server() as (read, write):
await server.run(read, write, server.create_initialization_options())
def main() -> None:
if "--list" in sys.argv: # quick offline self-check
for n, info in _AGENTS.items():
print(f"{n:22s} params: {', '.join(info['params']) or '-'}")
print(f"\n{len(_AGENTS)} agent tools + {len(MANAGEMENT_TOOLS)} management tools")
return
asyncio.run(_amain())
if __name__ == "__main__":
main()