| """Command line entry point for the Nexum runtime.""" |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import base64 |
| import hmac |
| import json |
| import os |
| from dataclasses import replace |
| from pathlib import Path |
| from typing import Any |
|
|
| from .bundle import release_artifact_sha256, validate_bundle |
| from .capabilities import capability_catalog |
| from .executor import ( |
| TOOL_CALL_END, |
| TOOL_CALL_START, |
| execute_tool_calls, |
| execute_tool_text, |
| list_tools, |
| parse_tool_calls, |
| tool_schemas, |
| ) |
| from .harness import NNFXHarnessRequest, NNFXHarnessSession |
| from .identity import candidate_identity_sha256, runtime_source_sha256 |
| from .runner import NexumLocalRunner |
| from .server import serve |
| from .status import SELF_CORRECTION_STATUS, SELF_IMPROVEMENT_STATUS, runtime_health |
| from .templates import TEMPLATES, render_template, template_names |
| from .tooling.artifacts import ArtifactStore |
| from .tooling.events import EventLog |
| from .tooling.language_packs import ( |
| analyze_language_file, |
| language_pack_catalog, |
| verify_language_packs, |
| ) |
| from .tooling.security import ApprovalStore |
| from .tooling.tasks import TaskStore |
|
|
|
|
| def _print_json(payload: object) -> None: |
| print(json.dumps(payload, indent=2, sort_keys=True)) |
|
|
|
|
| def _cmd_doctor(args: argparse.Namespace) -> int: |
| model_dir = str(args.model or os.environ.get("NEXUM_MODEL_DIR", "")).strip() |
| if not model_dir: |
| payload = runtime_health() |
| payload["error"] = "model directory is required" |
| _print_json(payload) |
| return 1 |
| report = validate_bundle(model_dir, deep=True) |
| payload = runtime_health(bundle_ok=report.ok) |
| payload["bundle"] = report.to_dict() |
| _print_json(payload) |
| return 0 if report.ok else 1 |
|
|
|
|
| def _cmd_bundle_validate(args: argparse.Namespace) -> int: |
| report = validate_bundle(args.model, deep=args.deep) |
| _print_json(report.to_dict()) |
| return 0 if report.ok else 1 |
|
|
|
|
| def _cmd_tools_list(_: argparse.Namespace) -> int: |
| _print_json({"tools": list_tools()}) |
| return 0 |
|
|
|
|
| def _cmd_tools_execute(args: argparse.Namespace) -> int: |
| metadata_supplied = bool( |
| args.call_id or args.idempotency_key or args.approval_id |
| ) |
| if metadata_supplied: |
| try: |
| parsed = parse_tool_calls(args.tool_text) |
| except ValueError as exc: |
| _print_json({"ok": False, "error": f"invalid tool call: {exc}"}) |
| return 1 |
| if len(parsed) != 1: |
| _print_json( |
| { |
| "ok": False, |
| "error": "execution metadata requires exactly one tool call", |
| } |
| ) |
| return 1 |
| call = replace( |
| parsed[0], |
| call_id=args.call_id, |
| idempotency_key=args.idempotency_key, |
| approval_id=args.approval_id, |
| ) |
| executed = execute_tool_calls( |
| (call,), |
| cwd=args.workspace, |
| timeout_s=args.timeout_s, |
| session_id=args.session_id, |
| ) |
| else: |
| executed = tuple( |
| execute_tool_text( |
| args.tool_text, |
| cwd=args.workspace, |
| timeout_s=args.timeout_s, |
| session_id=args.session_id, |
| ) |
| ) |
| results = [ |
| item.to_dict() |
| for item in executed |
| ] |
| ok = bool(results) and all(item["ok"] for item in results) |
| _print_json({"ok": ok, "results": results}) |
| return 0 if ok else 1 |
|
|
|
|
| def _cmd_tools_exec_smoke(args: argparse.Namespace) -> int: |
| workspace = Path(args.workspace).expanduser().resolve() |
| workspace.mkdir(parents=True, exist_ok=True) |
| call = f"{TOOL_CALL_START}[Bash(command='printf nexum-tool-ok')]{TOOL_CALL_END}" |
| results = [ |
| item.to_dict() |
| for item in execute_tool_text( |
| call, cwd=str(workspace), timeout_s=args.timeout_s |
| ) |
| ] |
| ok = ( |
| len(results) == 1 |
| and bool(results[0]["ok"]) |
| and "nexum-tool-ok" in str(results[0]["output"]) |
| ) |
| _print_json({"ok": ok, "results": results}) |
| return 0 if ok else 1 |
|
|
|
|
| def _cmd_status_self_correction(_: argparse.Namespace) -> int: |
| _print_json(SELF_CORRECTION_STATUS) |
| return 0 |
|
|
|
|
| def _cmd_status_self_improvement(_: argparse.Namespace) -> int: |
| _print_json(SELF_IMPROVEMENT_STATUS) |
| return 0 |
|
|
|
|
| def _configure_learning_state(args: argparse.Namespace) -> None: |
| if str(args.state_dir).strip(): |
| os.environ["NEXUM_STATE_DIR"] = str( |
| Path(args.state_dir).expanduser().resolve() |
| ) |
| os.environ["NEXUM_STATE_NAMESPACE"] = str(args.state_namespace) |
|
|
|
|
| def _validated_candidate_identity(model_dir: str) -> str: |
| normalized_model_dir = str(model_dir).strip() |
| if not normalized_model_dir: |
| raise ValueError("model directory is required") |
| report = validate_bundle(normalized_model_dir, deep=True) |
| if not report.ok: |
| raise RuntimeError("Nexum model bundle validation failed") |
| release_sha256 = release_artifact_sha256( |
| normalized_model_dir, |
| validated_report=report, |
| ) |
| return candidate_identity_sha256( |
| release_sha256, |
| runtime_source_sha256(), |
| ) |
|
|
|
|
| def _cmd_learning_status(args: argparse.Namespace) -> int: |
| _configure_learning_state(args) |
| candidate_sha256 = _validated_candidate_identity(args.model) |
| runner = NexumLocalRunner(args.model, device=args.device) |
| _print_json( |
| { |
| **runner.learning_status(), |
| "candidate_identity_sha256": candidate_sha256, |
| "state_namespace": runner.state_namespace, |
| } |
| ) |
| return 0 |
|
|
|
|
| def _cmd_learning_rollback(args: argparse.Namespace) -> int: |
| _configure_learning_state(args) |
| candidate_sha256 = _validated_candidate_identity(args.model) |
| if not hmac.compare_digest(args.expected_candidate, candidate_sha256): |
| raise ValueError("candidate identity changed before rollback") |
| runner = NexumLocalRunner(args.model, device=args.device) |
| _print_json( |
| { |
| **runner.rollback_learning( |
| target_generation=args.target_generation, |
| expected_generation=args.expected_generation, |
| ), |
| "candidate_identity_sha256": candidate_sha256, |
| "state_namespace": runner.state_namespace, |
| } |
| ) |
| return 0 |
|
|
|
|
| def _cmd_capabilities(_: argparse.Namespace) -> int: |
| _print_json(capability_catalog()) |
| return 0 |
|
|
|
|
| def _cmd_language_packs(args: argparse.Namespace) -> int: |
| if bool(args.verify): |
| verification = verify_language_packs() |
| _print_json(verification) |
| return 0 if bool(verification["ok"]) else 1 |
| _print_json(language_pack_catalog(args.query)) |
| return 0 |
|
|
|
|
| def _cmd_language_inspect(args: argparse.Namespace) -> int: |
| workspace = Path(args.workspace).expanduser().resolve() |
| target = (workspace / args.path).resolve() |
| try: |
| relative = target.relative_to(workspace) |
| except ValueError as exc: |
| raise ValueError("language inspection path leaves workspace") from exc |
| if relative.parts and relative.parts[0] == ".nexum": |
| raise ValueError("workspace control paths are not source inputs") |
| if not target.is_file(): |
| raise ValueError("language inspection target is not a file") |
| _print_json(analyze_language_file(target, language=args.language)) |
| return 0 |
|
|
|
|
| def _cmd_approvals_list(args: argparse.Namespace) -> int: |
| records = [ |
| record.__dict__ |
| for record in ApprovalStore(args.workspace).list( |
| session_id=args.session_id |
| ) |
| ] |
| _print_json({"approvals": records}) |
| return 0 |
|
|
|
|
| def _cmd_approvals_decide(args: argparse.Namespace) -> int: |
| record = ApprovalStore(args.workspace).decide( |
| args.approval_id, |
| approved=args.decision == "approve", |
| session_id=args.session_id, |
| ) |
| EventLog(args.workspace).append( |
| "approval_decision", |
| session_id=args.session_id, |
| status=record.status, |
| detail={"approval_id": record.approval_id, "tool": record.tool}, |
| ) |
| _print_json({"ok": True, "approval": record.__dict__}) |
| return 0 |
|
|
|
|
| def _cmd_tasks_list(args: argparse.Namespace) -> int: |
| records = [ |
| record.to_dict() |
| for record in TaskStore(args.workspace).list(session_id=args.session_id) |
| ] |
| _print_json({"tasks": records}) |
| return 0 |
|
|
|
|
| def _cmd_tasks_status(args: argparse.Namespace) -> int: |
| _print_json( |
| { |
| "task": TaskStore(args.workspace) |
| .status(args.task_id, session_id=args.session_id) |
| .to_dict() |
| } |
| ) |
| return 0 |
|
|
|
|
| def _cmd_tasks_cancel(args: argparse.Namespace) -> int: |
| task = TaskStore(args.workspace).cancel( |
| args.task_id, session_id=args.session_id |
| ) |
| _print_json({"ok": task.status == "cancelled", "task": task.to_dict()}) |
| return 0 if task.status == "cancelled" else 1 |
|
|
|
|
| def _cmd_artifacts_list(args: argparse.Namespace) -> int: |
| records = [ |
| record.to_dict() |
| for record in ArtifactStore(args.workspace).list( |
| session_id=args.session_id |
| ) |
| ] |
| _print_json({"artifacts": records}) |
| return 0 |
|
|
|
|
| def _cmd_artifacts_read(args: argparse.Namespace) -> int: |
| record, data = ArtifactStore(args.workspace).read( |
| args.artifact_id, |
| offset=args.offset, |
| length=args.length, |
| session_id=args.session_id, |
| ) |
| _print_json( |
| { |
| "artifact": record.to_dict(), |
| "offset": args.offset, |
| "data_base64": base64.b64encode(data).decode("ascii"), |
| } |
| ) |
| return 0 |
|
|
|
|
| def _cmd_events_list(args: argparse.Namespace) -> int: |
| events = [ |
| event.to_dict() |
| for event in EventLog(args.workspace).read( |
| session_id=args.session_id, |
| after_sequence=args.after, |
| ) |
| ] |
| _print_json({"events": events}) |
| return 0 |
|
|
|
|
| def _cmd_templates_list(_: argparse.Namespace) -> int: |
| _print_json({"templates": template_names()}) |
| return 0 |
|
|
|
|
| def _cmd_templates_show(args: argparse.Namespace) -> int: |
| _print_json({"name": args.name, "template": TEMPLATES[args.name]}) |
| return 0 |
|
|
|
|
| def _cmd_templates_write(args: argparse.Namespace) -> int: |
| out = Path(args.out).expanduser() |
| out.mkdir(parents=True, exist_ok=True) |
| for name in template_names(): |
| (out / f"{name}.txt").write_text(render_template(name), encoding="utf-8") |
| _print_json({"ok": True, "out": str(out), "templates": template_names()}) |
| return 0 |
|
|
|
|
| def _cmd_serve(args: argparse.Namespace) -> int: |
| serve( |
| host=args.host, |
| port=args.port, |
| model_dir=args.model, |
| workspace=args.workspace, |
| device=args.device, |
| api_key=os.environ.get("NEXUM_API_KEY", ""), |
| enable_tools=bool(args.enable_tools), |
| state_dir=args.state_dir, |
| state_namespace=args.state_namespace, |
| ) |
| return 0 |
|
|
|
|
| def _json_file(path: str) -> Any: |
| return json.loads(Path(path).expanduser().read_text(encoding="utf-8")) |
|
|
|
|
| def _completion_payload(args: argparse.Namespace) -> dict[str, Any]: |
| payload: dict[str, Any] = { |
| "model": "Nexum", |
| "temperature": float(args.temperature), |
| "session_id": str(args.session_id or ""), |
| } |
| if args.messages: |
| messages = _json_file(args.messages) |
| if not isinstance(messages, list): |
| raise ValueError("messages file must contain a JSON list") |
| payload["messages"] = messages |
| else: |
| payload["prompt"] = str(args.prompt or "") |
| if args.tools: |
| payload["tools"] = tool_schemas() |
| if args.max_completion_tokens is not None: |
| payload["max_completion_tokens"] = int(args.max_completion_tokens) |
| return payload |
|
|
|
|
| def _cmd_chat(args: argparse.Namespace) -> int: |
| response = NexumLocalRunner(args.model, device=args.device).chat( |
| _completion_payload(args) |
| ) |
| _print_json(response) |
| return 0 |
|
|
|
|
| def _cmd_agent(args: argparse.Namespace) -> int: |
| runner = NexumLocalRunner(args.model, device=args.device) |
| payload = _completion_payload(args) |
| payload["workspace"] = args.workspace |
| payload["timeout_s"] = float(args.timeout_s) |
| request = NNFXHarnessRequest.from_json(payload) |
| response = NNFXHarnessSession( |
| complete=runner.chat, |
| sign_observation=runner.sign_observation, |
| ).run(request) |
| _print_json(response.to_dict()) |
| return 0 if response.ok else 1 |
|
|
|
|
| def _add_completion_arguments( |
| parser: argparse.ArgumentParser, *, include_workspace: bool |
| ) -> None: |
| parser.add_argument( |
| "--model", default=os.environ.get("NEXUM_MODEL_DIR", ""), required=False |
| ) |
| parser.add_argument("--device", default=os.environ.get("NEXUM_DEVICE", "cuda:0")) |
| source = parser.add_mutually_exclusive_group(required=True) |
| source.add_argument("--prompt") |
| source.add_argument("--messages", help="JSON file containing chat messages") |
| parser.add_argument( |
| "--tools", action="store_true", help="include the built-in tool schemas" |
| ) |
| parser.add_argument("--temperature", type=float, default=0.0) |
| parser.add_argument("--max-completion-tokens", type=int) |
| parser.add_argument("--session-id", default="") |
| if include_workspace: |
| parser.add_argument( |
| "--workspace", default=os.environ.get("NEXUM_WORKSPACE", ".") |
| ) |
| parser.add_argument("--timeout-s", type=float, default=0.0) |
|
|
|
|
| def build_parser() -> argparse.ArgumentParser: |
| parser = argparse.ArgumentParser(prog="nexum", description="Nexum private runtime") |
| sub = parser.add_subparsers(dest="command", required=True) |
|
|
| doctor = sub.add_parser("doctor", help="show runtime health") |
| doctor.add_argument("--model", default=os.environ.get("NEXUM_MODEL_DIR", "")) |
| doctor.set_defaults(fn=_cmd_doctor) |
|
|
| bundle = sub.add_parser("bundle", help="model folder operations") |
| bundle_sub = bundle.add_subparsers(dest="bundle_command", required=True) |
| bundle_validate = bundle_sub.add_parser( |
| "validate", help="validate a flat model folder" |
| ) |
| bundle_validate.add_argument("--model", required=True) |
| bundle_validate.add_argument("--deep", action="store_true") |
| bundle_validate.set_defaults(fn=_cmd_bundle_validate) |
|
|
| tools = sub.add_parser("tools", help="tool operations") |
| tools_sub = tools.add_subparsers(dest="tools_command", required=True) |
| tools_list = tools_sub.add_parser("list", help="list tools") |
| tools_list.set_defaults(fn=_cmd_tools_list) |
| tools_execute = tools_sub.add_parser("execute", help="execute supplied tool text") |
| tools_execute.add_argument("tool_text") |
| tools_execute.add_argument("--workspace", default=".") |
| tools_execute.add_argument("--timeout-s", type=float, default=0.0) |
| tools_execute.add_argument("--session-id", default="") |
| tools_execute.add_argument( |
| "--call-id", default="", help="stable identifier for one exact tool call" |
| ) |
| tools_execute.add_argument( |
| "--idempotency-key", |
| default="", |
| help="caller-selected exactly-once key for one tool call", |
| ) |
| tools_execute.add_argument( |
| "--approval-id", |
| default="", |
| help="approved exact-action receipt used by one tool-call retry", |
| ) |
| tools_execute.set_defaults(fn=_cmd_tools_execute) |
| tools_smoke = tools_sub.add_parser( |
| "exec-smoke", help="run a small terminal tool smoke" |
| ) |
| tools_smoke.add_argument("--workspace", default=".") |
| tools_smoke.add_argument("--timeout-s", type=float, default=0.0) |
| tools_smoke.set_defaults(fn=_cmd_tools_exec_smoke) |
|
|
| status = sub.add_parser("status", help="runtime feature status") |
| status_sub = status.add_subparsers(dest="status_command", required=True) |
| sc = status_sub.add_parser("self-correction", help="show self-correction status") |
| sc.set_defaults(fn=_cmd_status_self_correction) |
| si = status_sub.add_parser("self-improvement", help="show self-improvement status") |
| si.set_defaults(fn=_cmd_status_self_improvement) |
|
|
| learning = sub.add_parser("learning", help="inspect or restore shared learning") |
| learning_sub = learning.add_subparsers(dest="learning_command", required=True) |
| for learning_parser in ( |
| learning_sub.add_parser("status", help="show validated learning history"), |
| learning_sub.add_parser("rollback", help="restore a retained generation"), |
| ): |
| learning_parser.add_argument( |
| "--model", |
| default=os.environ.get("NEXUM_MODEL_DIR", ""), |
| required=False, |
| ) |
| learning_parser.add_argument( |
| "--device", |
| default=os.environ.get("NEXUM_DEVICE", "cuda:0"), |
| ) |
| learning_parser.add_argument( |
| "--state-dir", |
| default=os.environ.get("NEXUM_STATE_DIR", ""), |
| ) |
| learning_parser.add_argument( |
| "--state-namespace", |
| default=os.environ.get("NEXUM_STATE_NAMESPACE", "default"), |
| ) |
| learning_status = learning_sub.choices["status"] |
| learning_status.set_defaults(fn=_cmd_learning_status) |
| learning_rollback = learning_sub.choices["rollback"] |
| learning_rollback.add_argument("--target-generation", type=int, required=True) |
| learning_rollback.add_argument("--expected-generation", type=int, required=True) |
| learning_rollback.add_argument("--expected-candidate", required=True) |
| learning_rollback.set_defaults(fn=_cmd_learning_rollback) |
|
|
| capabilities_parser = sub.add_parser( |
| "capabilities", help="show runtime capability discovery" |
| ) |
| capabilities_parser.set_defaults(fn=_cmd_capabilities) |
|
|
| language_packs_parser = sub.add_parser( |
| "language-packs", |
| help="show source-intelligence packs and measured native toolchains", |
| ) |
| language_packs_parser.add_argument( |
| "--query", |
| default="", |
| help="filter by language name, alias, identifier, or extension", |
| ) |
| language_packs_parser.add_argument( |
| "--verify", |
| action="store_true", |
| help="exercise every declared source-intelligence backend", |
| ) |
| language_packs_parser.set_defaults(fn=_cmd_language_packs) |
|
|
| language_inspect_parser = sub.add_parser( |
| "language-inspect", |
| help="inspect one workspace source file with its language pack", |
| ) |
| language_inspect_parser.add_argument("path") |
| language_inspect_parser.add_argument("--language", default="") |
| language_inspect_parser.add_argument("--workspace", default=".") |
| language_inspect_parser.set_defaults(fn=_cmd_language_inspect) |
|
|
| approvals = sub.add_parser("approvals", help="manage exact action approvals") |
| approvals_sub = approvals.add_subparsers(dest="approvals_command", required=True) |
| approvals_list = approvals_sub.add_parser("list", help="list approval records") |
| approvals_list.add_argument("--workspace", default=".") |
| approvals_list.add_argument("--session-id", default="") |
| approvals_list.set_defaults(fn=_cmd_approvals_list) |
| approvals_decide = approvals_sub.add_parser( |
| "decide", help="approve or deny one exact action" |
| ) |
| approvals_decide.add_argument("approval_id") |
| approvals_decide.add_argument("decision", choices=("approve", "deny")) |
| approvals_decide.add_argument("--session-id", required=True) |
| approvals_decide.add_argument("--workspace", default=".") |
| approvals_decide.set_defaults(fn=_cmd_approvals_decide) |
|
|
| tasks = sub.add_parser("tasks", help="inspect durable task lifecycle") |
| tasks_sub = tasks.add_subparsers(dest="tasks_command", required=True) |
| tasks_list = tasks_sub.add_parser("list", help="list durable tasks") |
| tasks_list.add_argument("--workspace", default=".") |
| tasks_list.add_argument("--session-id", default="") |
| tasks_list.set_defaults(fn=_cmd_tasks_list) |
| tasks_status = tasks_sub.add_parser("status", help="show one durable task") |
| tasks_status.add_argument("task_id") |
| tasks_status.add_argument("--workspace", default=".") |
| tasks_status.add_argument("--session-id", default="") |
| tasks_status.set_defaults(fn=_cmd_tasks_status) |
| tasks_cancel = tasks_sub.add_parser("cancel", help="cancel one running task") |
| tasks_cancel.add_argument("task_id") |
| tasks_cancel.add_argument("--workspace", default=".") |
| tasks_cancel.add_argument("--session-id", default="") |
| tasks_cancel.set_defaults(fn=_cmd_tasks_cancel) |
|
|
| artifacts = sub.add_parser("artifacts", help="inspect complete tool artifacts") |
| artifacts_sub = artifacts.add_subparsers(dest="artifacts_command", required=True) |
| artifacts_list = artifacts_sub.add_parser("list", help="list artifacts") |
| artifacts_list.add_argument("--workspace", default=".") |
| artifacts_list.add_argument("--session-id", default="") |
| artifacts_list.set_defaults(fn=_cmd_artifacts_list) |
| artifacts_read = artifacts_sub.add_parser("read", help="read an artifact range") |
| artifacts_read.add_argument("artifact_id") |
| artifacts_read.add_argument("--workspace", default=".") |
| artifacts_read.add_argument("--session-id", default="") |
| artifacts_read.add_argument("--offset", type=int, default=0) |
| artifacts_read.add_argument("--length", type=int) |
| artifacts_read.set_defaults(fn=_cmd_artifacts_read) |
|
|
| events = sub.add_parser("events", help="inspect ordered runtime receipts") |
| events.add_argument("--workspace", default=".") |
| events.add_argument("--session-id", default="") |
| events.add_argument("--after", type=int, default=0) |
| events.set_defaults(fn=_cmd_events_list) |
|
|
| templates = sub.add_parser("templates", help="template operations") |
| templates_sub = templates.add_subparsers(dest="templates_command", required=True) |
| templates_list = templates_sub.add_parser("list", help="list templates") |
| templates_list.set_defaults(fn=_cmd_templates_list) |
| templates_show = templates_sub.add_parser("show", help="show a template") |
| templates_show.add_argument("name", choices=template_names()) |
| templates_show.set_defaults(fn=_cmd_templates_show) |
| templates_write = templates_sub.add_parser( |
| "write", help="write templates to a folder" |
| ) |
| templates_write.add_argument("--out", required=True) |
| templates_write.set_defaults(fn=_cmd_templates_write) |
|
|
| chat = sub.add_parser("chat", help="run one Nexum completion") |
| _add_completion_arguments(chat, include_workspace=False) |
| chat.set_defaults(fn=_cmd_chat) |
|
|
| agent = sub.add_parser("agent", help="run the model-driven NNF X tool loop") |
| _add_completion_arguments(agent, include_workspace=True) |
| agent.set_defaults(fn=_cmd_agent) |
|
|
| server = sub.add_parser("serve", help="run HTTP server") |
| server.add_argument("--host", default="127.0.0.1") |
| server.add_argument("--port", type=int, default=8080) |
| server.add_argument("--model", default=os.environ.get("NEXUM_MODEL_DIR", "")) |
| server.add_argument("--workspace", default=os.environ.get("NEXUM_WORKSPACE", ".")) |
| server.add_argument("--device", default=os.environ.get("NEXUM_DEVICE", "cuda:0")) |
| server.add_argument("--state-dir", default=os.environ.get("NEXUM_STATE_DIR", "")) |
| server.add_argument( |
| "--state-namespace", |
| default=os.environ.get("NEXUM_STATE_NAMESPACE", "default"), |
| ) |
| server.add_argument( |
| "--enable-tools", |
| action="store_true", |
| default=os.environ.get("NEXUM_ENABLE_TOOLS", "").lower() |
| in {"1", "true", "yes"}, |
| help="allow authenticated HTTP tool execution and agent endpoints", |
| ) |
| server.set_defaults(fn=_cmd_serve) |
|
|
| return parser |
|
|
|
|
| def main(argv: list[str] | None = None) -> int: |
| args = build_parser().parse_args(argv) |
| return int(args.fn(args)) |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
