Skip to content

Instantly share code, notes, and snippets.

@kierr

kierr/always_allow_all.py

Created August 7, 2025 08:13
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save kierr/fa012d8c437fac80cab8fafb8b817ff4 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save kierr/fa012d8c437fac80cab8fafb8b817ff4 to your computer and use it in GitHub Desktop.
Download ZIP
Always Allow All (Kilo Code MCP Servers)
Raw
always_allow_all.py
#!/usr/bin/env python3
"""
always_allow_all.py [OPTIONS] CONFIG.json
Kilo Code currently has no way to always allow all tools for an MCP server.
This script will always allow every tool for every server in a provided configuration file.
Options:
-v, --verbose Increase output verbosity
-r, --retries Number of retries for flaky servers (default: 1)
-h, --help Show this help message
"""
from __future__ import annotations
import json, os, pathlib, subprocess, sys, textwrap, time, typing as t, itertools, argparse
try:
import requests # pip install requests if missing
except ModuleNotFoundError:
sys.exit("❌ Please `pip install requests` first")
# ───────────────────────── helpers ───────────────────────── #
JSON = t.Dict[str, t.Any]
RPC_REQ = {"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}}
def tools_from_http(url: str, quiet: bool = True) -> list[str]:
"""POST the tools/list request to an HTTP-exposed MCP server."""
if not quiet:
print(f"📡 Connecting to HTTP server: {url}")
try:
r = requests.post(url, json=RPC_REQ, timeout=15)
r.raise_for_status()
result = r.json()["result"]
names = [t["name"] for t in result["tools"]]
# Handle pagination if present
while result.get("nextCursor"):
rpc = RPC_REQ | {"params": {"cursor": result["nextCursor"]}}
r = requests.post(url, json=rpc, timeout=15)
r.raise_for_status()
result = r.json()["result"]
names.extend(t["name"] for t in result["tools"])
return names
except requests.exceptions.RequestException as e:
raise Exception(f"HTTP connection failed: {e}")
def tools_from_stdio(command: str, args: list[str], env: JSON, quiet: bool = True) -> list[str]:
"""Spawn the server (docker run / npx … ) and speak JSON-RPC over stdio."""
if not quiet:
cmd_display = f"{command} {' '.join(args)}"
print(f"🚀 Spawning server: {cmd_display}")
# Suppress server logs by redirecting stderr
stderr_redirect = subprocess.DEVNULL if quiet else None
proc = subprocess.Popen(
[command, *args],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=stderr_redirect,
text=True,
env={**os.environ, **env},
)
try:
proc.stdin.write(json.dumps(RPC_REQ) + "\n")
proc.stdin.flush()
raw = proc.stdout.readline().strip()
result = json.loads(raw)["result"]
names = [t["name"] for t in result["tools"]]
while result.get("nextCursor"):
rpc = RPC_REQ | {"params": {"cursor": result["nextCursor"]}}
proc.stdin.write(json.dumps(rpc) + "\n")
proc.stdin.flush()
raw = proc.stdout.readline().strip()
result = json.loads(raw)["result"]
names.extend(t["name"] for t in result["tools"])
return names
finally:
proc.terminate()
try: proc.wait(timeout=2)
except subprocess.TimeoutExpired:
proc.kill()
def enumerate_tools(entry: JSON, quiet: bool = True) -> list[str]:
"""Choose the correct transport automatically."""
if "url" in entry: # simplest case
return tools_from_http(entry["url"], quiet)
if entry.get("command") and entry.get("args"):
return tools_from_stdio(entry["command"], entry["args"], entry.get("env", {}), quiet)
raise ValueError("Cannot determine how to talk to server entry")
# ───────────────────────── main flow ───────────────────────── #
def main():
parser = argparse.ArgumentParser(
description="Always allow all tools for MCP servers in a configuration file",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=__doc__
)
parser.add_argument("config", help="Path to the MCP configuration JSON file")
parser.add_argument("-v", "--verbose", action="store_true", help="Increase output verbosity")
parser.add_argument("-r", "--retries", type=int, default=1, help="Number of retries for flaky servers (default: 1)")
try:
args = parser.parse_args()
except SystemExit:
# argparse calls sys.exit() on -h/--help, but we want to show our custom help
sys.exit(0)
path = pathlib.Path(args.config).expanduser().resolve()
if not path.exists():
sys.exit(f"❌ Configuration file not found: {path}")
try:
cfg: JSON = json.loads(path.read_text())
except json.JSONDecodeError as e:
sys.exit(f"❌ Invalid JSON in configuration file: {e}")
if args.verbose:
print(f"📁 Loading configuration from: {path}")
# Collect statistics
stats = {
"total_servers": 0,
"processed_servers": 0,
"successful_servers": 0,
"failed_servers": [],
"disabled_servers": [],
"total_tools_added": 0,
"total_tools_existing": 0,
"server_details": {}
}
mcp_servers = cfg.get("mcpServers", {})
stats["total_servers"] = len(mcp_servers)
if args.verbose:
print(f"🔧 Found {stats['total_servers']} MCP servers")
for name, entry in mcp_servers.items():
if entry.get("disabled", False):
stats["disabled_servers"].append(name)
if args.verbose:
print(f"⏭️ {name}: skipped (disabled)")
continue
stats["processed_servers"] += 1
success = False
last_error = None
tools = []
# Try multiple times for flaky servers
for attempt in range(args.retries):
try:
if args.retries > 1 and args.verbose:
print(f"🔄 {name}: attempt {attempt + 1}/{args.retries}")
tools = enumerate_tools(entry, not args.verbose)
success = True
break
except Exception as e:
last_error = e
if attempt < args.retries - 1: # Not the last attempt
if args.verbose:
print(f"⚠️ {name}: attempt {attempt + 1} failed, retrying...")
time.sleep(1) # Brief delay before retry
if not success:
stats["failed_servers"].append((name, str(last_error)))
if args.verbose:
print(f"❌ {name}: {last_error} – skipped")
continue
stats["successful_servers"] += 1
existing = set(entry.get("alwaysAllow", []))
merged = sorted(existing | set(tools))
# Store details for summary
stats["server_details"][name] = {
"existing_count": len(existing),
"new_count": len(tools),
"total_after_update": len(merged),
"added_count": len(merged) - len(existing)
}
stats["total_tools_added"] += len(merged) - len(existing)
stats["total_tools_existing"] += len(existing)
if merged != entry.get("alwaysAllow"):
entry["alwaysAllow"] = merged
added = len(merged) - len(existing)
if args.verbose:
print(f"✅ {name}: added {added} tool(s), now {len(merged)} total")
elif args.verbose:
print(f"✅ {name}: already up to date ({len(merged)} tools)")
# Print summary
print(f"📊 Summary:")
print(f" Total servers found: {stats['total_servers']}")
print(f" Processed servers: {stats['processed_servers']}")
print(f" Successful servers: {stats['successful_servers']}")
print(f" Disabled servers: {len(stats['disabled_servers'])}")
print(f" Failed servers: {len(stats['failed_servers'])}")
if stats["disabled_servers"]:
print(f" Disabled: {', '.join(stats['disabled_servers'])}")
if stats["failed_servers"]:
print(f" Failed:")
for name, error in stats["failed_servers"]:
print(f" - {name}: {error}")
print(f" Tools added: {stats['total_tools_added']}")
print(f" Tools already present: {stats['total_tools_existing']}")
# Print per-server details
if stats["server_details"]:
print(f" Server details:")
for name, details in stats["server_details"].items():
if details["added_count"] > 0:
print(f" - {name}: {details['existing_count']} existing, {details['added_count']} added, {details['total_after_update']} total")
else:
print(f" - {name}: {details['existing_count']} existing, up to date")
if stats["total_tools_added"] > 0:
path.write_text(json.dumps(cfg, indent=2) + "\n")
print(f"✅ Configuration updated with {stats['total_tools_added']} new tools")
else:
print("✅ No changes needed - all tools already allowed")
if __name__ == "__main__":
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment