让你的 Agent 远程运维 RunPod 和 Railway 上的程序

先说结论

• 拆成三个平面。重启、回滚、看日志、改环境变量，走平台控制面（API / CLI / MCP）就够，不需要 shell；进容器排查才用数据面（SSH），而且默认只读；代码层面的修复走 PR 和自动部署，别在容器里热改，下次部署就没了。
• harness 留在你的可信环境里（本机、自有服务器或一台小 VPS），RunPod 和 Railway 只是“工具目标”。API token 和 SSH 私钥只存在于 harness 进程，不进模型上下文，也不放进远端容器。OpenAI 在 2026 年 4 月的 Agents SDK 更新里强调的“harness 与 compute 分离”是同一个思路。
• Railway 今年对 agent 很友好。官方 remote MCP（mcp.railway.com，OAuth 登录）；railway ssh -- <命令> 可以非交互执行，也支持 scp / sftp；GraphQL API 有只作用于单个环境的 project token；还能用 railway agent -p "…" --json 把 Railway 自己的 agent 当子 agent 调。
• RunPod 的控制面在迁移。新集成直接用 REST v2（api.runpod.io/v2），pod 的 start / stop / restart / terminate 统一走一个 action 接口，v1 年底前下线。数据面要用 Full SSH（公网 IP 加暴露 22/tcp），代理式的基础 SSH 不支持 SCP / SFTP。stop 会清空 container disk，只保留 /workspace。
• 权限在 harness 端强制，不靠 prompt。读操作自动执行；变更类（重启、改变量、部署）要人审批，或者只在 staging 自动；破坏性操作（terminate、删卷、删服务）干脆不注册成工具。
• 日志和命令输出都是不可信输入，是间接 prompt injection 的入口。输出要截断、脱敏，高风险工具不能因为“日志里的一句话”被触发。

把“远程维护”拆成三个平面

让 agent 去远端维护程序，要做的事按“走什么通道、动的是什么”分成三类。分开设计的原因很实际：三类的凭据、风险和审批规则都不一样，全塞进一个 bash 工具就没法管。

五种接入架构

这五种不互斥。比较稳的组合是“远端即工具目标”加“只走 GitOps”打底，GPU 上的重调试再加“agent 驻留远端”。

可以参照 OpenAI 给出的对比：Claude Agent SDK 的常见用法是把 harness 连同 agent loop 一起放进沙箱；新版 OpenAI Agents SDK 则把 harness 留在可信运行时，沙箱只是它调用的执行面。远程运维更适合后者，因为 token 和审批逻辑不该和被维护的程序住在一起。

Railway：接口很全，权限要自己收

Railway 在 2026 年把 agent 当成一等用户，CLI、hosted MCP、agent skills、云端 agent、SSH 入口都齐了。对自研 harness 最有用的是下面几块。

通过 SSH 在容器里改的文件，下次部署就没了（挂载的 volume 除外）。修复要回写到代码或配置里，所以数据面只用来排查和临时止血。

RunPod：控制面在迁移，SSH 要选对

RunPod 有两种形态。Pods 是持久的 GPU / CPU 实例，可以 SSH 进去维护；Serverless 的 worker 是短暂的，维护对象是 endpoint 配置、镜像和环境变量，而不是某台机器。

给 agent 的原则：Pods 上的改动只落在 /workspace 或 network volume；环境依赖的变化要回写到镜像或模板，否则下一次 stop 就回到原样。

Fly.io、Render、Modal 和自有 VPS 也基本是“API / CLI / MCP 加 SSH”两条路，套用同一套工具层即可。自有 VPS 或裸机可以再加 Teleport 这类方案，用短期证书替代长期 SSH key，并录制会话。

工具怎么设计

给模型一个万能 bash 最省事，也最难管。更好的做法是把常用动作做成 typed tools，参数是 service、env、pod_id 这类 harness 能校验的枚举值，再留一个受限的 remote_exec 当逃生口。每个工具标上平面和风险等级，策略引擎按等级处理。

• 参数用枚举。service 只能是白名单里的几个名字，pod_id 只能是登记过的 ID。模型写错或被诱导时，校验先拦下。
• 返回要短。日志默认只给最后几百行，支持按关键字过滤；超过上限就截断，并告诉模型省略了多少。
• 变更类工具返回“变更后的状态”，而不是“请求成功”。重启后自动轮询到健康或超时，再把结果交给模型。

安全清单

• 凭据只在 harness。token 和 SSH 私钥由 harness 进程持有，模型看不到，远端容器里也不放。给 agent 单独的 token 和 key，可以随时吊销。
• 最小权限。Railway 优先 project token（只作用于一个环境）；RunPod 用 Restricted 或 Read Only key；MCP 优先 OAuth，本地不落 token 文件。
• 目标白名单。只允许登记过的 service、environment、pod ID；生产和 staging 用不同的策略。Railway 的 workspace SSH key 能进该工作区所有服务，所以把 agent 管的服务单独放一个工作区。
• 审批。变更类操作走 Slack 或命令行审批；破坏性操作不注册为工具，由人去控制台操作。
• 执行约束。每条命令都有超时；输出截断；含 shell 元字符（; | & $ ` > <）的命令一律走审批；路径黑名单覆盖 .env、~/.ssh、credentials 之类。
• 输出脱敏。结果回到模型之前，把 token、私钥块、数据库连接串打码。
• 把日志当作不可信输入。工具输出是间接 prompt injection 的主要入口之一；高风险动作只能由策略放行，不能由输出里的文字触发。
• 审计。每次工具调用写一行 JSONL：时间、工具、参数、决策、退出码、输出哈希。自管的 VPS 或裸机可以用短期证书加会话录制的方案。
• 复核与幂等。Railway 的 mutation 不保证 exactly-once；RunPod 变更后用 GET 复核。
• 节流。Railway API 有每小时配额，agent 轮询要限速。

代码骨架

下面的骨架演示工具注册、风险分级、目标白名单、审批、审计和脱敏怎么串起来。SSH 用 asyncssh，HTTP 用 httpx。Railway 的日志、重启、回滚等具体 query / mutation 名称，请在 GraphiQL（railway.com/graphiql）里用 introspection 确认，这里只放官方文档里出现过的查询。

# remote_ops.py：自研 harness 的远程运维工具层骨架（Python 3.10+，示意用）
# pip install asyncssh httpx
import hashlib, json, os, re, time
from dataclasses import dataclass
from enum import Enum
from typing import Any, Awaitable, Callable

import asyncssh
import httpx


class Risk(Enum):
    READ = "read"        # 自动执行
    CHANGE = "change"    # 需要人工批准（staging 可放宽）
    # 破坏性操作（terminate、删卷、删服务）不定义成工具


# ---------- 数据面：SSH ----------
@dataclass
class SSHTarget:
    host: str        # Railway: ssh.railway.com；RunPod: pod 公网 IP
    port: int        # RunPod: 22/tcp 映射出的公网端口；Railway: 22
    username: str    # Railway: 服务域名或 service instance ID；RunPod: root
    key_path: str    # 专给 agent 的私钥，只存在于 harness 所在机器
    # 固定 host key，不要关掉校验；新 pod 首次连接先把 host key 写进这个文件
    known_hosts: str = os.path.expanduser("~/.ssh/known_hosts_agent")

    async def run(self, cmd: str, timeout: float = 60) -> dict:
        async with asyncssh.connect(
            self.host, port=self.port, username=self.username,
            client_keys=[self.key_path], known_hosts=self.known_hosts,
        ) as conn:
            r = await conn.run(cmd, check=False, timeout=timeout)
            return {"exit": r.exit_status, "stdout": r.stdout, "stderr": r.stderr}

# Railway 若用库直连遇到兼容问题，可以改成子进程调用：
#   railway ssh -s <服务> -e <环境> -- <命令>


# ---------- 控制面：Railway GraphQL ----------
class Railway:
    URL = "https://backboard.railway.com/graphql/v2"

    def __init__(self, token: str, project_token: bool = True):
        # project token 用 Project-Access-Token 头；account / workspace token 用 Bearer
        self.headers = ({"Project-Access-Token": token} if project_token
                        else {"Authorization": f"Bearer {token}"})

    async def gql(self, query: str, variables: dict | None = None) -> dict:
        async with httpx.AsyncClient(timeout=30) as c:
            r = await c.post(self.URL, headers=self.headers,
                             json={"query": query, "variables": variables or {}})
        r.raise_for_status()
        body = r.json()
        if body.get("errors"):       # 鉴权失败也可能是 HTTP 200 + errors
            raise RuntimeError(body["errors"])
        return body["data"]

    async def token_scope(self) -> dict:
        # 官方文档里的示例查询；其余字段名以 GraphiQL introspection 为准
        return await self.gql("query { projectToken { projectId environmentId } }")


# ---------- 控制面：RunPod REST v2 ----------
class RunPod:
    BASE = "https://api.runpod.io/v2"

    def __init__(self, api_key: str):
        self.headers = {"Authorization": f"Bearer {api_key}"}

    async def get_pod(self, pod_id: str) -> dict:
        async with httpx.AsyncClient(timeout=30) as c:
            r = await c.get(f"{self.BASE}/pods/{pod_id}", headers=self.headers)
        r.raise_for_status()
        return r.json()

    async def pod_action(self, pod_id: str, action: str) -> dict:
        if action not in {"start", "stop", "restart"}:   # terminate 永远不交给模型
            raise ValueError(f"action {action!r} not allowed")
        async with httpx.AsyncClient(timeout=30) as c:
            r = await c.post(f"{self.BASE}/pods/{pod_id}/action",
                             headers=self.headers, json={"action": action})
        r.raise_for_status()
        # 别只信 200：真实实现里应轮询到目标状态或超时，再返回
        return await self.get_pod(pod_id)


# ---------- 工具、策略、审计 ----------
@dataclass
class Tool:
    name: str
    risk: Risk
    fn: Callable[..., Awaitable[Any]]
    description: str
    schema: dict     # JSON Schema，转换成你所用模型 API 的工具定义

SAFE_PREFIX = ("nvidia-smi", "df -h", "free -m", "uptime", "ps aux",
               "ls ", "tail -n ", "journalctl -n ")   # 故意不放 cat
BLOCKED = (".env", ".ssh", "credentials", "id_rsa", "id_ed25519", "secret")
SHELL_META = re.compile(r"[;&|`$<>\\\n]")
SECRET = re.compile(
    r"sk-[A-Za-z0-9_-]{16,}|rpa_[A-Za-z0-9]{16,}|postgres(?:ql)?://\S+"
    r"|-----BEGIN [A-Z ]*PRIVATE KEY-----[\s\S]+?-----END [A-Z ]*PRIVATE KEY-----")


def exec_needs_approval(cmd: str) -> bool:
    return (bool(SHELL_META.search(cmd)) or not cmd.startswith(SAFE_PREFIX)
            or any(b in cmd for b in BLOCKED))


def redact(text: str, limit: int = 20_000) -> str:
    text = SECRET.sub("[REDACTED]", text or "")
    if len(text) <= limit:
        return text
    return f"[前面省略 {len(text) - limit} 个字符]\n" + text[-limit:]


@dataclass
class Policy:
    targets: dict[str, SSHTarget]    # 登记过的远端目标
    pods: set[str]                   # 允许操作的 RunPod pod ID
    approve: Callable[[str, dict], Awaitable[bool]]   # 例如发到 Slack 等人确认
    audit_path: str = "audit.jsonl"

    async def allow(self, tool: Tool, args: dict) -> bool:
        if "target" in args and args["target"] not in self.targets:
            return False
        if "pod_id" in args and args["pod_id"] not in self.pods:
            return False
        needs = tool.risk is Risk.CHANGE or (
            tool.name == "remote_exec" and exec_needs_approval(args.get("cmd", "")))
        return await self.approve(tool.name, args) if needs else True

    def audit(self, tool: str, args: dict, allowed: bool, output: str = "") -> None:
        rec = {"ts": time.time(), "tool": tool, "args": args, "allowed": allowed,
               "sha256": hashlib.sha256(output.encode()).hexdigest()}
        with open(self.audit_path, "a", encoding="utf-8") as f:
            f.write(json.dumps(rec, ensure_ascii=False) + "\n")


async def dispatch(tools: dict[str, Tool], policy: Policy, name: str, args: dict) -> str:
    # 生产环境先用 jsonschema 按 tool.schema 校验 args，拒绝多余字段
    tool = tools.get(name)
    if tool is None:
        return "error: unknown tool"
    if not await policy.allow(tool, args):
        policy.audit(name, args, False)
        return "denied by policy"     # 明确告诉模型被拒，别静默失败
    try:
        out = json.dumps(await tool.fn(**args), ensure_ascii=False, default=str)
    except Exception as e:            # 超时、连不上也要回给模型
        out = f"error: {type(e).__name__}: {e}"
    policy.audit(name, args, True, out)
    return redact(out)


def build_tools(policy: Policy, runpod: RunPod) -> dict[str, Tool]:
    async def remote_exec(target: str, cmd: str):
        return await policy.targets[target].run(cmd, timeout=60)

    async def pod_status(pod_id: str):
        return await runpod.get_pod(pod_id)

    async def pod_restart(pod_id: str):
        return await runpod.pod_action(pod_id, "restart")

    pod_arg = {"type": "object", "required": ["pod_id"], "additionalProperties": False,
               "properties": {"pod_id": {"type": "string", "enum": sorted(policy.pods)}}}
    tools = [
        Tool("remote_exec", Risk.READ, remote_exec,
             "在登记过的远端容器里执行一条命令。只读诊断命令自动执行，其余需要人工批准。",
             {"type": "object", "required": ["target", "cmd"], "additionalProperties": False,
              "properties": {"target": {"type": "string", "enum": sorted(policy.targets)},
                             "cmd": {"type": "string", "maxLength": 500}}}),
        Tool("pod_status", Risk.READ, pod_status, "查看 RunPod pod 的状态与配置。", pod_arg),
        Tool("pod_restart", Risk.CHANGE, pod_restart,
             "重启 RunPod pod。container disk 会被清空，只有 /workspace 保留。", pod_arg),
    ]
    return {t.name: t for t in tools}

接到你的 loop 里：把每个 Tool.schema 转成所用模型 API 的工具定义；模型发出 tool call 时调用 dispatch()，把返回的字符串作为 tool result 送回去。approve 可以接 Slack 按钮、命令行确认或内部审批系统。Railway 的日志、重启、回滚按同样方式包成工具即可。

落地顺序

1. 只读起步。接 get_status、tail_logs 和指标，只发只读凭据（Railway project token、RunPod Read Only key）。先让 agent 学会“看”。
2. 加变更工具和审批。restart、redeploy、rollback、set_env 全部走审批，staging 可以放开自动。
3. 打开数据面。Full SSH 或 railway ssh，白名单只读命令自动，其余审批；截断和脱敏要在这一步之前做好。
4. 打通修复闭环。在沙箱或 staging 复现，提 PR，CI 通过后自动部署，看日志和健康检查验证，失败就回滚。
5. 值班化。用平台 webhook 或监控告警触发 agent；harness 要能断点续跑（事件日志持久化），审批改成异步（比如 Slack 按钮）；把排障手册写成 skills 或 AGENTS.md。

沙箱在这里的位置

E2B、Daytona、Modal、Vercel Sandbox、Cloudflare Sandboxes、Runloop 这类沙箱平台，在远程运维里不是维护对象，而是车间。

• 复现和验证。先在隔离环境复现 bug、跑修复和测试，再提 PR。Railway Sandboxes 能走项目私网连到同项目的服务，适合复现“只在 Railway 上出现”的问题，最好连 staging。
• 放驻留 agent。需要把 coding agent 放到远端时，放进沙箱比直接进生产容器安全。
• 现成的抽象层。OpenAI Agents SDK 在 2026 年 4 月的更新里内置了 Blaxel、Cloudflare、Daytona、E2B、Modal、Runloop、Vercel 的沙箱支持，也允许自带沙箱，并能在容器失效后从快照恢复继续跑。Claude Managed Agents 的 self-hosted environment 可以把工具执行放到你自己的机器，或者 E2B、Modal、Vercel、Fly.io、AWS Lambda MicroVMs 等平台上。