Files
stackpilot/backend/routers/agents.py
T
menzeljandClaude Opus 4.8 012614f5fb Phase 13: multi-host networks & images (0.14.0)
Networks and Images are now per-host, rendered as a section for the local host
plus one per registered agent (like the Stacks page).

- agent_app.py: new /agent/networks (list/inspect/containers/connect/disconnect/
  create/delete/prune) and /agent/images (list/updates/check), reusing
  network_service and a new image_service; DockerError mapped to HTTP status
  (forbidden -> 400 so the proxy doesn't treat it as a token failure).
- routers/agents.py: proxy routes at /api/agents/{id}/networks/* and
  /api/agents/{id}/images/*, audit-logging mutations.
- services/image_service.py: extracted the image-listing logic so the central
  router and the agent share it.
- Frontend: networksApi/imagesApi take an optional agentId; Networks/Images
  pages render NetworksSection/ImagesSection per host with a shared HostHeader.
  Remote "Prune unused" networks resolves the address-pool-exhaustion deploy
  error from the UI.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-08 11:52:53 +00:00

645 lines
20 KiB
Python

"""Remote host (agent) management + proxied stack/system operations."""
from __future__ import annotations
import asyncio
import os
import tempfile
from fastapi import APIRouter, Depends, File, Form, HTTPException, Query, Request, UploadFile
from fastapi.responses import FileResponse
from pydantic import BaseModel
from sqlmodel import Session, select
from starlette.background import BackgroundTask
from auth import get_current_user, require_admin
from database import get_session
from models.agent import Agent, AgentCreate, AgentRead, AgentUpdate
from models.backup_destination import BackupDestination
from models.stack import StackCreate, StackUpdate
from models.user import User
from routers.networks import ContainerRef, NetworkCreate
from services import (
agent_service,
audit_service,
backup_destination_service as dest_service,
backup_service,
compose_service,
)
from services.agent_service import AgentError
router = APIRouter(prefix="/api/agents", tags=["agents"])
_ACTIONS = {"start", "stop", "restart", "pull", "update", "down"}
class AgentPushBody(BaseModel):
destination_id: int
include_volumes: bool = True
stop_first: bool = True
class AgentRestoreFromBody(BaseModel):
destination_id: int
name: str
target_id: str | None = None
overwrite: bool = False
restore_volumes: bool = True
def _get_dest(session: Session, dest_id: int) -> BackupDestination:
d = session.get(BackupDestination, dest_id)
if not d:
raise HTTPException(status_code=404, detail=f"Destination {dest_id} not found")
return d
def _ip(request: Request) -> str:
return request.client.host if request.client else "unknown"
def _to_read(a: Agent) -> AgentRead:
return AgentRead(
id=a.id,
name=a.name,
url=a.url,
status=a.status,
hostname=a.hostname,
last_seen=a.last_seen,
created_at=a.created_at,
token_set=bool(a.token),
)
def _get_or_404(session: Session, agent_id: int) -> Agent:
agent = session.get(Agent, agent_id)
if not agent:
raise HTTPException(status_code=404, detail=f"Agent {agent_id} not found")
return agent
def _raise(exc: AgentError):
raise HTTPException(
status_code=exc.status if exc.status >= 400 else 502,
detail={"error": exc.error, "detail": exc.detail},
)
async def _proxy(session: Session, agent: Agent, method: str, path: str, **kw):
try:
return await agent_service.call(session, agent, method, path, **kw)
except AgentError as exc:
_raise(exc)
# --------------------------------------------------------------------------- #
# CRUD
# --------------------------------------------------------------------------- #
@router.get("", response_model=list[AgentRead])
async def list_agents(
refresh: bool = Query(True),
session: Session = Depends(get_session),
_user: User = Depends(get_current_user),
) -> list[AgentRead]:
agents = session.exec(select(Agent).order_by(Agent.id)).all()
if refresh:
for agent in agents:
await agent_service.ping(session, agent)
return [_to_read(a) for a in agents]
@router.post("", response_model=AgentRead, status_code=201)
async def create_agent(
body: AgentCreate,
request: Request,
session: Session = Depends(get_session),
user: User = Depends(require_admin),
) -> AgentRead:
agent = Agent(name=body.name, url=body.url.rstrip("/"), token=body.token)
session.add(agent)
session.commit()
session.refresh(agent)
# Validate connectivity immediately (best-effort; agent is saved regardless).
await agent_service.ping(session, agent)
audit_service.record(
session, user=user.username, action="agent.create", target=agent.name,
detail=agent.url, ip=_ip(request),
)
return _to_read(agent)
@router.put("/{agent_id}", response_model=AgentRead)
async def update_agent(
agent_id: int,
body: AgentUpdate,
request: Request,
session: Session = Depends(get_session),
user: User = Depends(require_admin),
) -> AgentRead:
agent = _get_or_404(session, agent_id)
if body.name is not None:
agent.name = body.name
if body.url is not None:
agent.url = body.url.rstrip("/")
if body.token:
agent.token = body.token
agent.status = "unknown"
session.add(agent)
session.commit()
session.refresh(agent)
await agent_service.ping(session, agent)
audit_service.record(
session, user=user.username, action="agent.update", target=agent.name,
ip=_ip(request),
)
return _to_read(agent)
@router.delete("/{agent_id}")
def delete_agent(
agent_id: int,
request: Request,
session: Session = Depends(get_session),
user: User = Depends(require_admin),
) -> dict:
agent = _get_or_404(session, agent_id)
name = agent.name
session.delete(agent)
session.commit()
audit_service.record(
session, user=user.username, action="agent.delete", target=name, ip=_ip(request),
)
return {"ok": True}
@router.post("/{agent_id}/ping")
async def ping_agent(
agent_id: int,
session: Session = Depends(get_session),
_user: User = Depends(get_current_user),
) -> dict:
agent = _get_or_404(session, agent_id)
return await agent_service.ping(session, agent)
# --------------------------------------------------------------------------- #
# Proxied stack + system operations
# --------------------------------------------------------------------------- #
@router.get("/{agent_id}/system")
async def agent_system(
agent_id: int,
session: Session = Depends(get_session),
_user: User = Depends(get_current_user),
):
agent = _get_or_404(session, agent_id)
return await _proxy(session, agent, "GET", "/agent/system")
@router.get("/{agent_id}/stacks")
async def agent_stacks(
agent_id: int,
session: Session = Depends(get_session),
_user: User = Depends(get_current_user),
) -> list[dict]:
agent = _get_or_404(session, agent_id)
stacks = await _proxy(session, agent, "GET", "/agent/stacks") or []
for s in stacks:
s["agent_id"] = agent.id
s["agent_name"] = agent.name
return stacks
@router.get("/{agent_id}/stacks/{stack_id}")
async def agent_stack_detail(
agent_id: int,
stack_id: str,
session: Session = Depends(get_session),
_user: User = Depends(get_current_user),
) -> dict:
agent = _get_or_404(session, agent_id)
data = await _proxy(session, agent, "GET", f"/agent/stacks/{stack_id}")
data["agent_id"] = agent.id
data["agent_name"] = agent.name
return data
@router.get("/{agent_id}/stacks/{stack_id}/logs")
async def agent_stack_logs(
agent_id: int,
stack_id: str,
tail: int = Query(200, le=2000),
session: Session = Depends(get_session),
_user: User = Depends(get_current_user),
) -> dict:
agent = _get_or_404(session, agent_id)
return await _proxy(
session, agent, "GET", f"/agent/stacks/{stack_id}/logs", params={"tail": tail}
)
@router.post("/{agent_id}/stacks", status_code=201)
async def agent_create_stack(
agent_id: int,
body: StackCreate,
request: Request,
session: Session = Depends(get_session),
user: User = Depends(require_admin),
) -> dict:
agent = _get_or_404(session, agent_id)
result = await _proxy(
session, agent, "POST", "/agent/stacks",
json={"name": body.name, "yaml": body.yaml, "env": body.env},
)
audit_service.record(
session, user=user.username, action="agent.stack.create",
target=f"{agent.name}/{result.get('id')}", ip=_ip(request),
)
return result
@router.put("/{agent_id}/stacks/{stack_id}")
async def agent_update_stack(
agent_id: int,
stack_id: str,
body: StackUpdate,
request: Request,
session: Session = Depends(get_session),
user: User = Depends(require_admin),
) -> dict:
agent = _get_or_404(session, agent_id)
result = await _proxy(
session, agent, "PUT", f"/agent/stacks/{stack_id}",
json={"yaml": body.yaml, "env": body.env},
)
audit_service.record(
session, user=user.username, action="agent.stack.update",
target=f"{agent.name}/{stack_id}", ip=_ip(request),
)
return result
@router.delete("/{agent_id}/stacks/{stack_id}")
async def agent_delete_stack(
agent_id: int,
stack_id: str,
request: Request,
delete_files: bool = Query(True),
session: Session = Depends(get_session),
user: User = Depends(require_admin),
) -> dict:
agent = _get_or_404(session, agent_id)
result = await _proxy(
session, agent, "DELETE", f"/agent/stacks/{stack_id}",
params={"delete_files": delete_files},
)
audit_service.record(
session, user=user.username, action="agent.stack.delete",
target=f"{agent.name}/{stack_id}", ip=_ip(request),
)
return result
@router.post("/{agent_id}/stacks/{stack_id}/{action}")
async def agent_lifecycle(
agent_id: int,
stack_id: str,
action: str,
request: Request,
session: Session = Depends(get_session),
user: User = Depends(require_admin),
) -> dict:
if action not in _ACTIONS:
raise HTTPException(status_code=400, detail=f"Unknown action '{action}'")
agent = _get_or_404(session, agent_id)
result = await _proxy(session, agent, "POST", f"/agent/stacks/{stack_id}/{action}")
audit_service.record(
session, user=user.username, action=f"agent.stack.{action}",
target=f"{agent.name}/{stack_id}", ip=_ip(request),
)
return result
# --------------------------------------------------------------------------- #
# Backup / restore of remote stacks (streamed agent <-> main <-> destination)
# --------------------------------------------------------------------------- #
@router.get("/{agent_id}/stacks/{stack_id}/backup")
async def agent_backup_download(
agent_id: int,
stack_id: str,
request: Request,
include_volumes: bool = Query(True),
stop_first: bool = Query(True),
session: Session = Depends(get_session),
user: User = Depends(require_admin),
):
agent = _get_or_404(session, agent_id)
tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".tar.gz")
tmp.close()
try:
await agent_service.download_to_file(
session, agent, f"/agent/stacks/{stack_id}/backup", tmp.name,
params={"include_volumes": include_volumes, "stop_first": stop_first},
)
except AgentError as exc:
if os.path.exists(tmp.name):
os.unlink(tmp.name)
_raise(exc)
audit_service.record(
session, user=user.username, action="agent.stack.backup",
target=f"{agent.name}/{stack_id}", ip=_ip(request),
)
fname = backup_service.backup_filename(
stack_id, include_volumes, prefix=compose_service.slugify(agent.name)
)
return FileResponse(
tmp.name, media_type="application/gzip", filename=fname,
background=BackgroundTask(os.unlink, tmp.name),
)
@router.post("/{agent_id}/stacks/{stack_id}/backup/push")
async def agent_backup_push(
agent_id: int,
stack_id: str,
body: AgentPushBody,
request: Request,
session: Session = Depends(get_session),
user: User = Depends(require_admin),
) -> dict:
agent = _get_or_404(session, agent_id)
dest = _get_dest(session, body.destination_id)
tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".tar.gz")
tmp.close()
try:
try:
await agent_service.download_to_file(
session, agent, f"/agent/stacks/{stack_id}/backup", tmp.name,
params={"include_volumes": body.include_volumes, "stop_first": body.stop_first},
)
except AgentError as exc:
_raise(exc)
fname = backup_service.backup_filename(
stack_id, body.include_volumes, prefix=compose_service.slugify(agent.name)
)
try:
await asyncio.to_thread(dest_service.upload, dest, tmp.name, fname)
except dest_service.DestinationError as exc:
raise HTTPException(status_code=502, detail=str(exc)) from exc
finally:
if os.path.exists(tmp.name):
os.unlink(tmp.name)
audit_service.record(
session, user=user.username, action="agent.stack.backup.push",
target=f"{agent.name}/{stack_id}", detail=f"{dest.name}:{fname}", ip=_ip(request),
)
return {"ok": True, "destination": dest.name, "name": fname}
@router.post("/{agent_id}/stacks/restore")
async def agent_restore_upload(
agent_id: int,
request: Request,
file: UploadFile = File(...),
target_id: str | None = Form(None),
overwrite: bool = Form(False),
restore_volumes: bool = Form(True),
session: Session = Depends(get_session),
user: User = Depends(require_admin),
) -> dict:
agent = _get_or_404(session, agent_id)
tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".tar.gz")
try:
while chunk := await file.read(1024 * 1024):
tmp.write(chunk)
tmp.close()
result = await agent_service.upload_file(
session, agent, "/agent/stacks/restore", tmp.name,
file.filename or "backup.tar.gz",
{
"target_id": target_id or "",
"overwrite": str(overwrite).lower(),
"restore_volumes": str(restore_volumes).lower(),
},
)
except AgentError as exc:
_raise(exc)
finally:
if os.path.exists(tmp.name):
os.unlink(tmp.name)
audit_service.record(
session, user=user.username, action="agent.stack.restore",
target=f"{agent.name}/{result.get('stack_id')}", ip=_ip(request),
)
return result
@router.post("/{agent_id}/stacks/restore-from")
async def agent_restore_from(
agent_id: int,
body: AgentRestoreFromBody,
request: Request,
session: Session = Depends(get_session),
user: User = Depends(require_admin),
) -> dict:
agent = _get_or_404(session, agent_id)
dest = _get_dest(session, body.destination_id)
tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".tar.gz")
tmp.close()
try:
try:
await asyncio.to_thread(dest_service.download, dest, body.name, tmp.name)
except dest_service.DestinationError as exc:
raise HTTPException(status_code=502, detail=str(exc)) from exc
result = await agent_service.upload_file(
session, agent, "/agent/stacks/restore", tmp.name, body.name,
{
"target_id": body.target_id or "",
"overwrite": str(body.overwrite).lower(),
"restore_volumes": str(body.restore_volumes).lower(),
},
)
except AgentError as exc:
_raise(exc)
finally:
if os.path.exists(tmp.name):
os.unlink(tmp.name)
audit_service.record(
session, user=user.username, action="agent.stack.restore",
target=f"{agent.name}/{result.get('stack_id')}", detail=f"from {dest.name}:{body.name}",
ip=_ip(request),
)
return result
# --------------------------------------------------------------------------- #
# Networks (proxied)
# --------------------------------------------------------------------------- #
@router.get("/{agent_id}/networks")
async def agent_networks(
agent_id: int,
session: Session = Depends(get_session),
_user: User = Depends(get_current_user),
) -> list[dict]:
agent = _get_or_404(session, agent_id)
return await _proxy(session, agent, "GET", "/agent/networks") or []
@router.get("/{agent_id}/networks/{network_id}")
async def agent_network_inspect(
agent_id: int,
network_id: str,
session: Session = Depends(get_session),
_user: User = Depends(get_current_user),
) -> dict:
agent = _get_or_404(session, agent_id)
return await _proxy(session, agent, "GET", f"/agent/networks/{network_id}")
@router.get("/{agent_id}/networks/{network_id}/containers")
async def agent_network_containers(
agent_id: int,
network_id: str,
session: Session = Depends(get_session),
_user: User = Depends(get_current_user),
) -> list[dict]:
agent = _get_or_404(session, agent_id)
return await _proxy(session, agent, "GET", f"/agent/networks/{network_id}/containers") or []
@router.post("/{agent_id}/networks/prune")
async def agent_network_prune(
agent_id: int,
request: Request,
session: Session = Depends(get_session),
user: User = Depends(require_admin),
) -> dict:
agent = _get_or_404(session, agent_id)
result = await _proxy(session, agent, "POST", "/agent/networks/prune")
audit_service.record(
session, user=user.username, action="agent.network.prune", target=agent.name,
detail=str(result.get("NetworksDeleted") or []), ip=_ip(request),
)
return result
@router.post("/{agent_id}/networks", status_code=201)
async def agent_network_create(
agent_id: int,
body: NetworkCreate,
request: Request,
session: Session = Depends(get_session),
user: User = Depends(require_admin),
) -> dict:
agent = _get_or_404(session, agent_id)
result = await _proxy(session, agent, "POST", "/agent/networks", json=body.model_dump())
audit_service.record(
session, user=user.username, action="agent.network.create",
target=f"{agent.name}/{body.name}", ip=_ip(request),
)
return result
@router.delete("/{agent_id}/networks/{network_id}")
async def agent_network_delete(
agent_id: int,
network_id: str,
request: Request,
session: Session = Depends(get_session),
user: User = Depends(require_admin),
) -> dict:
agent = _get_or_404(session, agent_id)
result = await _proxy(session, agent, "DELETE", f"/agent/networks/{network_id}")
audit_service.record(
session, user=user.username, action="agent.network.delete",
target=f"{agent.name}/{network_id}", ip=_ip(request),
)
return result
@router.post("/{agent_id}/networks/{network_id}/connect")
async def agent_network_connect(
agent_id: int,
network_id: str,
body: ContainerRef,
request: Request,
session: Session = Depends(get_session),
user: User = Depends(require_admin),
) -> dict:
agent = _get_or_404(session, agent_id)
result = await _proxy(
session, agent, "POST", f"/agent/networks/{network_id}/connect", json=body.model_dump()
)
audit_service.record(
session, user=user.username, action="agent.network.connect",
target=f"{agent.name}/{network_id}", detail=body.container, ip=_ip(request),
)
return result
@router.post("/{agent_id}/networks/{network_id}/disconnect")
async def agent_network_disconnect(
agent_id: int,
network_id: str,
body: ContainerRef,
request: Request,
session: Session = Depends(get_session),
user: User = Depends(require_admin),
) -> dict:
agent = _get_or_404(session, agent_id)
result = await _proxy(
session, agent, "POST", f"/agent/networks/{network_id}/disconnect", json=body.model_dump()
)
audit_service.record(
session, user=user.username, action="agent.network.disconnect",
target=f"{agent.name}/{network_id}", detail=body.container, ip=_ip(request),
)
return result
# --------------------------------------------------------------------------- #
# Images (proxied)
# --------------------------------------------------------------------------- #
@router.get("/{agent_id}/images")
async def agent_images(
agent_id: int,
session: Session = Depends(get_session),
_user: User = Depends(get_current_user),
) -> list[dict]:
agent = _get_or_404(session, agent_id)
return await _proxy(session, agent, "GET", "/agent/images") or []
@router.get("/{agent_id}/images/updates")
async def agent_image_updates(
agent_id: int,
session: Session = Depends(get_session),
_user: User = Depends(get_current_user),
) -> dict:
agent = _get_or_404(session, agent_id)
return await _proxy(session, agent, "GET", "/agent/images/updates")
@router.post("/{agent_id}/images/check")
async def agent_image_check(
agent_id: int,
request: Request,
session: Session = Depends(get_session),
user: User = Depends(require_admin),
) -> dict:
agent = _get_or_404(session, agent_id)
result = await _proxy(session, agent, "POST", "/agent/images/check")
audit_service.record(
session, user=user.username, action="agent.image.check", target=agent.name,
ip=_ip(request),
)
return result