tail_job_logs

Instructions

Implementation Reference

• src/silentwatch_mcp/server.py:175-180 (handler)

  The call_tool handler for 'tail_job_logs' — extracts job_id and lines from arguments, calls backend.tail_logs(), wraps result in JobLogTail, and serializes via _serialize.

  if name == "tail_job_logs":
      job_id = arguments["job_id"]
      lines = int(arguments.get("lines", 50))
      log_lines = await backend.tail_logs(job_id, lines=lines)
      response = JobLogTail(job_id=job_id, lines=log_lines, total_lines_returned=len(log_lines))
      return _serialize(response)

• src/silentwatch_mcp/types.py:137-142 (schema)

  JobLogTail schema — the Pydantic response model for the tail_job_logs tool, with job_id, lines (list of strings), and total_lines_returned.

  class JobLogTail(BaseModel):
      """Response for `tail_job_logs`."""
  
      job_id: str
      lines: list[str]
      total_lines_returned: int

• src/silentwatch_mcp/server.py:120-132 (registration)

  Tool registration in list_tools() — declares the tool's name, description, and inputSchema (job_id required, lines optional default 50).

      Tool(
          name="tail_job_logs",
          description="Most recent N log lines for a job (newest last).",
          inputSchema={
              "type": "object",
              "properties": {
                  "job_id": {"type": "string"},
                  "lines": {"type": "integer", "default": 50},
              },
              "required": ["job_id"],
          },
      ),
  ]

• src/silentwatch_mcp/backends/base.py:37-44 (helper)

  Abstract tail_logs method in CronBackend base class — defines the contract all backends must implement.

  @abstractmethod
  async def tail_logs(self, job_id: str, lines: int = 50) -> list[str]:
      """Return up to `lines` recent log output lines for a job, newest last.
  
      For backends that don't maintain a separate log surface (e.g., mock or
      openclaw-jsonl which puts output in run records), implementations can
      synthesize this from `get_job_runs` outputs.
      """

• src/silentwatch_mcp/backends/mock.py:139-148 (helper)

  Mock backend's tail_logs implementation — synthesizes log lines from run records with timestamp, status, run_id, exit_code, and output snippet.

  async def tail_logs(self, job_id: str, lines: int = 50) -> list[str]:
      runs = self._runs.get(job_id, [])
      out: list[str] = []
      for run in runs:
          ts = run.started_at.isoformat()
          status = run.status.value
          out.append(f"{ts} [{status.upper()}] run_id={run.run_id} exit={run.exit_code}")
          if run.output_snippet:
              out.append(f"{ts} [OUTPUT] {run.output_snippet[:200]}")
      return out[-lines:]