vrr/Pulse
2
0
Fork 0
mirror of https://github.com/rcourtman/Pulse.git synced 2026-07-10 00:14:38 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

backend and governance: MCP contract, agent capabilities, API, and release-control

Browse source 
Manifest-backed MCP tools, prompts, and resources with surface affordance contracts; agent capability manifest and governance projection; API contract tests and capability route projection; operations-loop and intelligence-funnel telemetry; release-control subsystem documentation, registry, and tooling; licensing and configuration.
This commit is contained in:
rcourtman 2026-06-23 17:26:15 +01:00
parent 27ba85ddc6
commit ee8a24e14a
279 changed files with 43945 additions and 9404 deletions
Download patch file Download diff file Expand all files Collapse all files

265
cmd/pulse-mcp/README.md
View file

@ -1,14 +1,40 @@
# pulse-mcp
`pulse-mcp` is a Model Context Protocol (MCP) server that wraps Pulse's
agent surface as a tool set. It lets Claude Desktop, Claude Code, and any
other MCP-speaking client drive Pulse natively: list findings, read the
fleet, drill into a resource, set operator intent, run governed actions.
`pulse-mcp` is the Model Context Protocol (MCP) adapter for Pulse
Intelligence. It exposes the same governed operations substrate that
Pulse Assistant uses inside the app, but as tools for Claude Desktop,
Claude Code, OpenCode, and any other MCP-speaking client: list findings,
read the fleet, drill into a resource, set operator intent, and run
governed actions.
## Pulse Intelligence surface contract
<!-- pulse-mcp-surface-contract:start -->
- **Pulse Intelligence Core**: Canonical context, governed actions, safety gates, approval state, action audit, and verification shared by Pulse Assistant, Pulse MCP, and Pulse Patrol.
- **Pulse Patrol**: Patrol is the first-party operations surface: it watches, investigates, acts within the chosen Patrol mode, verifies outcomes, and records what happened.
- **Pulse Assistant** (the native Pulse surface): The contextual explanation, approval, and handoff surface for Patrol findings, governed actions, verification, and operator questions. Affordances: tools and interactive questions.
- **Pulse MCP**: The external-agent adapter that projects canonical Pulse Intelligence capabilities as MCP tools. Affordances: tools, resources, prompts, and capability metadata.
<!-- pulse-mcp-surface-contract:end -->
The adapter is manifest-driven. Every tool it exposes is one entry in
Pulse's hand-authored capabilities manifest at `/api/agent/capabilities`.
Adding a capability there extends this server automatically. There is no
hardcoded tool list to keep in sync.
Pulse's canonical capabilities manifest at `/api/agent/capabilities`.
Adding a request/response capability to the manifest-owned Pulse MCP surface
tool contract extends this server automatically. There is no hardcoded tool
list to keep in sync, and there is no separate MCP-only action path to maintain
beside Pulse Assistant. `tools/list` uses the shared Pulse Intelligence
projection to expose the manifest metadata in each tool description, including
route, auth scope, action mode (`read`, `mixed`, `write`), approval policy
(`scope_only`, `action_plan`), request/response shape, and stable error codes.
Operator-state writes plus governed action and finding lifecycle tools also
expose typed `inputSchema` definitions for their top-level arguments, including
stable enums for criticality, approval decisions, and dismissal reasons.
The same shared projection exposes `resources/list` and `resources/read`
when the manifest contains the canonical fleet and per-resource context
capabilities. Resource URIs use `pulse://resource/<resource-id>` and
`resources/read` returns the same JSON bundle as `get_resource_context`.
It also exposes `prompts/list` and `prompts/get` from the manifest-owned
`workflowPrompts` catalogue. Those prompts are workflow hints over the
manifest tools and resource URIs, not a separate MCP playbook layer.
## Install
@ -77,26 +103,42 @@ Use any of the install paths above. The rest of this guide assumes
### 2. Mint an API token
Pulse needs a token with `monitoring:read` for the read tools, and
`monitoring:write` if you want the write tools (`set_operator_state`,
`clear_operator_state`) to work. Mint one in **Settings → Security → API
Tokens**.
Pulse fetches the live capabilities manifest first, and every tool
description includes its required auth scope. For a read-only external
agent, start with `monitoring:read`. For the full published Pulse
Intelligence surface, mint a token with the current manifest scopes:
<!-- pulse-mcp-scope-list:start -->
`monitoring:read`, `monitoring:write`, `settings:read`, `settings:write`, and `ai:execute`
<!-- pulse-mcp-scope-list:end -->
You can also mint narrower tokens for specific workflows. For example,
omit `settings:write` if the client should not add or edit monitored
infrastructure sources, and omit `ai:execute` if it should not review
Patrol findings or plan, approve, or execute governed actions. Mint the
token in **Settings →
Security → API Tokens**.
### 3. Wire it into your client
#### Claude Desktop
<!-- pulse-mcp-client-config:start -->
Most MCP clients need the same manifest-owned runtime facts: server name `pulse`, command `pulse-mcp`, base URL flag `--base-url`, default URL `http://localhost:7655`, and token environment variable `PULSE_API_TOKEN`.
The generated examples below cover the currently declared config families: `OpenCode`, `Claude-style clients`, and `custom MCP clients`.
If your client uses a different outer config format, keep those runtime facts and adapt only the wrapper.
Edit `~/Library/Application Support/Claude/claude_desktop_config.json`
(macOS) or the platform equivalent. Add a `pulse` entry under
`mcpServers`:
#### OpenCode
Uses OpenCode's top-level mcp object.
Add this to `opencode.json`, `opencode.jsonc`, and `~/.config/opencode/opencode.json`:
```json
{
"mcpServers": {
"$schema": "https://opencode.ai/config.json",
"mcp": {
"pulse": {
"command": "/usr/local/bin/pulse-mcp",
"args": ["--base-url", "http://localhost:7655"],
"env": {
"type": "local",
"command": ["pulse-mcp", "--base-url", "http://localhost:7655"],
"enabled": true,
"environment": {
"PULSE_API_TOKEN": "your-token-here"
}
}
@ -104,11 +146,12 @@ Edit `~/Library/Application Support/Claude/claude_desktop_config.json`
}
```
Restart Claude Desktop. The Pulse tools appear in the tool picker.
Restart OpenCode or reload its config after saving. Pulse tools appear under the `pulse` MCP server name.
#### Claude Code
#### Claude-style clients
Add the server to your project's `.mcp.json` (or your user-level config):
Uses the common mcpServers object supported by Claude Desktop and Claude Code.
Use this shape for `Claude Desktop` and `Claude Code` in `~/Library/Application Support/Claude/claude_desktop_config.json` and `.mcp.json`:
```json
{
@ -124,6 +167,14 @@ Add the server to your project's `.mcp.json` (or your user-level config):
}
```
Restart your client after saving the config. If the client cannot resolve `pulse-mcp` from `PATH`, use the full binary path.
#### custom MCP clients
Keeps the Pulse MCP command, base URL flag, and token environment variable while adapting the outer client config shape.
Use command `pulse-mcp`, pass `--base-url http://localhost:7655`, set `PULSE_API_TOKEN` to the API token, and keep the server name `pulse` when the client asks for one.
<!-- pulse-mcp-client-config:end -->
## Configuration
| Flag | Default | Purpose |
@ -161,71 +212,88 @@ flag off and consume the SSE stream directly.
When `--emit-notifications` is on, the `initialize` response
advertises the supported event kinds under
`capabilities.experimental.pulseNotifications.kinds`. Clients that
don't understand the experimental block ignore it silently.
`capabilities.experimental.pulseNotifications.kinds`. That list is
projected from the shared Pulse Intelligence event vocabulary, so it
stays aligned with `/api/agent/events`. Clients that don't understand
the experimental block ignore it silently.
## What the tools do
The exact set is whatever your Pulse instance's manifest declares. As of
this writing, the published capabilities are:
The exact request/response tool set is whatever your Pulse instance's
manifest declares. The generated inventory below reflects the canonical
manifest in this checkout; run `tools/list` from your MCP client to see
the live set.
<!-- pulse-mcp-tools:start -->
**Context (read-only):**
- `get_resource_context` returns the situated picture of one resource:
identity, operator-set state, active findings, pending approvals,
recent actions including refused dispatches with their stable error
tokens.
- `get_fleet_context` returns a thin per-resource rollup across the org
for triage: identity, operator flags, per-severity finding counts,
pending-approval count.
- `get_resource_context` (Get resource context, `GET /api/agent/resource-context/{resourceId}`, scope `monitoring:read`, mode `read`, approval `scope_only`): Return the situated picture of a resource — identity, operator-set state with maintenance-window-active flag, active findings, pending approvals scoped to this resource, recent actions including refused dispatches. Command fields are redacted for monitoring-read API tokens unless the token also has ai:execute.
- `get_fleet_context` (Get fleet context, `GET /api/agent/fleet-context`, scope `monitoring:read`, mode `read`, approval `scope_only`): Return a thin per-resource triage rollup across every resource visible to the org — identity, operator flags (intentionallyOffline, neverAutoRemediate, maintenanceWindowActive), per-severity finding counts (total/critical/warning/info), and pending-approval count. One read for 'where do I focus?'; follow up via get_resource_context for depth.
- `get_operations_loop_status` (Get Patrol work status, `GET /api/agent/operations-loop/status`, scope `monitoring:read`, mode `read`, approval `scope_only`): Return the current content-safe Patrol work status: Patrol issue evidence, pending approvals, governed decisions/actions, verified outcomes, Patrol control outcome evidence, compatibility aliases, and optional MCP readiness. The payload is count-only and deliberately omits finding ids, action ids, prompts, commands, resource names, and output.
**Operator state (per-resource intent):**
- `get_operator_state` reads the operator-set state for a resource
(intentionally offline, never-auto-remediate, maintenance window,
criticality).
- `set_operator_state` replaces the entire record. The server populates
attribution (`setAt`, `setBy`) so client values cannot spoof it.
- `clear_operator_state` removes the entry. Idempotent.
- `get_operator_state` (Get operator state, `GET /api/resources/{resourceId}/operator-state`, scope `monitoring:read`, mode `read`, approval `scope_only`): Read the operator-set state for a resource (intentionally offline, never auto-remediate, maintenance window, criticality).
- `set_operator_state` (Set operator state, `PUT /api/resources/{resourceId}/operator-state`, scope `monitoring:write`, mode `write`, approval `scope_only`): Replace the operator-set state for a resource. URL canonicalId wins over body; server populates setAt and setBy from the authenticated identity.
- `clear_operator_state` (Clear operator state, `DELETE /api/resources/{resourceId}/operator-state`, scope `monitoring:write`, mode `write`, approval `scope_only`): Remove any operator-set state for a resource. Idempotent — succeeds whether or not an entry was present.
**Findings (Patrol lifecycle):**
- `list_findings` returns every Patrol finding (active, dismissed,
resolved). Filter client-side.
- `acknowledge_finding` marks a finding as seen but keeps it visible.
- `snooze_finding` hides a finding for a duration in hours.
- `dismiss_finding` permanently dismisses a finding with a reason
(`not_an_issue`, `expected_behavior`, `will_fix_later`).
- `resolve_finding` manually marks a finding resolved.
- `list_findings` (List findings, `GET /api/ai/patrol/findings`, scope `ai:execute`, mode `read`, approval `scope_only`): List all Patrol findings (active, dismissed, resolved). Filter client-side on returned shape.
- `acknowledge_finding` (Acknowledge finding, `POST /api/ai/patrol/acknowledge`, scope `ai:execute`, mode `write`, approval `scope_only`): Mark a finding as seen but keep it visible. Auto-resolves when the underlying condition clears.
- `snooze_finding` (Snooze finding, `POST /api/ai/patrol/snooze`, scope `ai:execute`, mode `write`, approval `scope_only`): Hide a finding for a defined duration in hours.
- `dismiss_finding` (Dismiss finding, `POST /api/ai/patrol/dismiss`, scope `ai:execute`, mode `write`, approval `scope_only`): Dismiss a finding with a reason: not_an_issue (permanent suppression), expected_behavior (acknowledged forever), or will_fix_later (7-day reminder commitment).
- `resolve_finding` (Resolve finding, `POST /api/ai/patrol/resolve`, scope `ai:execute`, mode `write`, approval `scope_only`): Manually mark a finding as resolved when the underlying issue has been fixed out-of-band.
**Actions (governed plan/approval/execute):**
- `plan_action` validates an action request, records the plan, and
returns approval policy, blast radius, plan hash, and preflight
summary.
- `decide_action` records an approval or rejection for a planned
action.
- `execute_action` executes a previously planned and approved action
and returns the persisted audit result.
- `plan_action` (Plan action, `POST /api/actions/plan`, scope `ai:execute`, mode `write`, approval `action_plan`): Plan an action against a resource. The planner validates the request, looks up the capability on the resource, checks executor-owned live availability, and returns an ActionPlan with the approval policy, blast radius, plan hash, and preflight summary. The plan is persisted to the audit history at the planned/pending state only after the live availability check passes, so subsequent decide_action and execute_action calls can reference it by id. Plan-and-execute is a two-step flow when the resulting plan requires approval, one-step otherwise.
- `decide_action` (Decide action, `POST /api/actions/{actionId}/decision`, scope `ai:execute`, mode `write`, approval `action_plan`): Record an approval decision (approved or rejected) on a previously planned action. The actor is taken from the authenticated identity; an explicit reason can be passed in the body. Idempotent on the persisted decision: re-deciding a non-pending action surfaces the action_not_pending stable code so agents can branch on the conflict rather than retrying blindly.
- `execute_action` (Execute action, `POST /api/actions/{actionId}/execute`, scope `ai:execute`, mode `write`, approval `action_plan`): Execute a previously planned and (when required) approved action. Returns the persisted audit record with the execution result attached. Refuses with stable codes when the action is in the wrong lifecycle state (action_not_approved, action_already_executing, action_execution_final, action_dry_run_only, action_plan_expired), when the approved plan no longer matches the current resource/capability contract (action_plan_drift), when the target is operator-locked against automated remediation (resource_remediation_locked), or when the API instance has no executor wired (action_executor_unavailable). action.completed SSE events fire on every terminal state so agents watching the stream do not need to poll this endpoint after dispatch.
**Provisioning (infrastructure onboarding):**
- `discover_lan` scans a subnet, or returns cached scan results, to
find candidate infrastructure hosts before import.
- `list_nodes` returns configured Pulse infrastructure sources with
credential secret values redacted.
- `add_node` adds a Proxmox VE, Proxmox Backup Server, or Proxmox Mail
Gateway source.
- `update_node` updates a configured source without changing omitted
fields.
- `remove_node` removes a configured source.
- `test_node_credentials` validates proposed source credentials without
saving them.
- `test_node_connection` validates a saved source by id.
- `refresh_node_cluster_membership` re-detects Proxmox VE cluster
endpoint metadata for an existing source.
- `list_nodes` (List nodes, `GET /api/config/nodes`, scope `settings:read`, mode `read`, approval `scope_only`): List configured infrastructure sources that Pulse can monitor or manage. Credential secret values are redacted; use the returned id with update_node, remove_node, test_node_connection, or refresh_node_cluster_membership.
- `add_node` (Add node, `POST /api/config/nodes`, scope `settings:write`, mode `write`, approval `scope_only`): Add a Proxmox VE, Proxmox Backup Server, or Proxmox Mail Gateway source to Pulse after credentials have been collected, generated, or approved.
- `update_node` (Update node, `PUT /api/config/nodes/{nodeId}`, scope `settings:write`, mode `write`, approval `scope_only`): Update a configured infrastructure source. Omitted fields preserve the current value; tokenValue or password only changes when supplied.
- `remove_node` (Remove node, `DELETE /api/config/nodes/{nodeId}`, scope `settings:write`, mode `write`, approval `scope_only`): Remove a configured infrastructure source from Pulse by node id.
- `test_node_credentials` (Test node credentials, `POST /api/config/nodes/test-config`, scope `settings:write`, mode `mixed`, approval `scope_only`): Validate proposed source credentials and connection details without saving them to Pulse.
- `test_node_connection` (Test node connection, `POST /api/config/nodes/{nodeId}/test`, scope `settings:write`, mode `mixed`, approval `scope_only`): Validate the saved connection for an existing configured infrastructure source.
- `refresh_node_cluster_membership` (Refresh node cluster membership, `POST /api/config/nodes/{nodeId}/refresh-cluster`, scope `settings:write`, mode `write`, approval `scope_only`): Re-detect Proxmox VE cluster membership and endpoint metadata for a configured source.
- `discover_lan` (Discover LAN, `POST /api/discover`, scope `settings:write`, mode `mixed`, approval `scope_only`): Scan a subnet, or return cached scan results, to find candidate infrastructure hosts before deciding which sources to add to Pulse.
<!-- pulse-mcp-tools:end -->
Run `tools/list` from your MCP client to see the live set.
## Workflow prompts
Clients that support MCP prompts can also discover Pulse workflow hints.
They are projected from the manifest-owned `workflowPrompts` catalogue:
<!-- pulse-mcp-prompts:start -->
- `pulse_triage_fleet` (Triage fleet): Triage the Pulse fleet using the canonical fleet context capability, then choose where deeper investigation is warranted.
- `pulse_operations_loop` (Ask Patrol to handle an issue): Have Patrol investigate active findings, follow the configured Patrol mode, take approved actions, verify the outcome, and record what happened.
- `pulse_investigate_resource` (Investigate resource; required argument `resourceId`): Investigate one Pulse resource using the canonical resource context capability and resource URI projection.
- `pulse_review_finding` (Review finding; required argument `finding_id`): Review one Patrol finding and propose the safest governed next step.
<!-- pulse-mcp-prompts:end -->
Run `prompts/list` from your MCP client to see the live prompt set.
## Resource browser
Clients that support MCP resources can call `resources/list` to browse
the same canonical resource IDs returned by `get_fleet_context`, then
`resources/read` to fetch the full `get_resource_context` JSON bundle
for one resource.
Example URI:
```text
pulse://resource/vm:101
```
The adapter does not maintain a separate resource registry. The resource
list is rebuilt through the live fleet-context capability, and each read
is a normal per-resource context capability call with the same auth scope,
redaction rules, and error envelope as the tool surface.
## Stable error envelope
@ -239,13 +307,24 @@ The MCP tool result wraps that JSON in a text content block with
`isError: true`. Branch on `error` (snake_case stable codes); use
`message` for surfacing to humans, never for branching.
Capability-specific codes the substrate currently emits include
`resource_not_found` (depth read on an unknown id),
`operator_state_not_set` (read on a resource with no operator entry), and
`operator_state_invalid` (write rejected by the validator). Cross-cutting
codes from the auth / multi-tenant middleware (`invalid_org`,
`org_suspended`, `access_denied`) can apply to any authenticated tool
call.
Capability-specific stable codes are advertised by the manifest:
<!-- pulse-mcp-errors:start -->
- `get_resource_context`: `resource_not_found`
- `get_operator_state`: `operator_state_not_set`
- `set_operator_state`: `operator_state_invalid`
- `acknowledge_finding`: `invalid_finding_request`, `finding_not_found`, `finding_action_not_allowed`, and `patrol_unavailable`
- `snooze_finding`: `invalid_finding_request`, `finding_not_found`, `finding_action_not_allowed`, and `patrol_unavailable`
- `dismiss_finding`: `invalid_finding_request`, `finding_not_found`, `finding_action_not_allowed`, and `patrol_unavailable`
- `resolve_finding`: `invalid_finding_request`, `finding_not_found`, `finding_action_not_allowed`, and `patrol_unavailable`
- `plan_action`: `invalid_action_request`, `resource_not_found`, `capability_not_found`, and `action_execution_unavailable`
- `decide_action`: `missing_id`, `invalid_id`, `invalid_action_decision`, `action_not_found`, `action_not_pending`, and `action_plan_expired`
- `execute_action`: `missing_id`, `invalid_id`, `invalid_action_execution`, `action_not_found`, `action_not_approved`, `action_already_executing`, `action_execution_final`, `action_dry_run_only`, `action_plan_expired`, `action_plan_drift`, `resource_remediation_locked`, and `action_executor_unavailable`
<!-- pulse-mcp-errors:end -->
Cross-cutting codes from the auth / multi-tenant middleware
(`invalid_org`, `org_suspended`, `access_denied`) can apply to any
authenticated tool call.
## Known limitations
@ -263,10 +342,6 @@ call.
lifetime. Restart `pulse-mcp` to pick up new capabilities after
upgrading Pulse.
- **No resource URIs.** MCP supports a `resources/` channel in addition
to `tools/`. The adapter exposes only tools today; this is sufficient
for the substrate's current surface and keeps the adapter small.
## Troubleshooting
**"env var PULSE_API_TOKEN is empty" on startup.**
@ -281,9 +356,13 @@ the proxy is gating the manifest endpoint. Make sure
`/api/agent/capabilities` is reachable without a credential, the same
way `/api/health` is.
**Tools work, but `set_operator_state` returns 403 access_denied.**
Your token is missing `monitoring:write`. Either mint a new token with
both scopes, or restrict yourself to the read tools.
**Tools work, but a write or action tool returns 403 access_denied.**
Your token is missing that tool's required scope. Run `tools/list` and
check the `required scope` line in the tool description. Operator-state
writes need `monitoring:write`, provisioning tools need `settings:read`
or `settings:write`, and Patrol finding plus governed action tools need
`ai:execute`. Mint a narrower or broader token depending on which
external-agent workflows you want to allow.
**Tools list is empty.**
The adapter filters `subscribe_events` out (it is not request/response
@ -293,16 +372,20 @@ manifest is empty, which is a Pulse-side bug; check
## Implementation notes
The adapter is one stdlib-only Go file (`main.go`) in around 430 lines.
It speaks JSON-RPC 2.0 over stdio with line-delimited framing, logs to
stderr to keep the JSON-RPC channel on stdout clean, and derives each
tool's input schema from the manifest entry: `{name}` segments in the
declared path become required string properties, and non-GET/DELETE
tools accept a free-form `body` object.
The adapter binary is intentionally thin. It speaks JSON-RPC 2.0 over
stdio with line-delimited framing, logs to stderr to keep the JSON-RPC
channel on stdout clean, and delegates method semantics, tool projection,
call-argument normalization, HTTP projection, and result envelopes to the
shared Pulse Intelligence contracts in `internal/agentcapabilities`.
Tool input schemas come from the manifest-owned JSON Schema when present;
the shared projection only falls back to deriving path placeholders and a
permissive body object for capabilities that have not yet authored a
stricter schema.
The companion worked example, `cmd/agent-probe`, walks the same
substrate as a plain HTTP client and is a useful reference for anyone
building a non-MCP integration. Together the two binaries demonstrate
the substrate's two consumer profiles: stdio MCP and HTTP API. Both are
manifest-driven and stdlib-only, so adding capabilities to Pulse extends
both without code changes here.
Pulse Intelligence's external-agent profiles: stdio MCP and direct HTTP
API. Pulse Assistant remains the first-party in-app surface over the
same governed contracts; `pulse-mcp` is the adapter for users who prefer
an external agent host.

601
cmd/pulse-mcp/main.go
View file

@ -1,11 +1,12 @@
// Command pulse-mcp is a minimal MCP (Model Context Protocol)
// adapter that wraps Pulse's agent surface for stdio-speaking
// clients like Claude Desktop and Claude Code. It is the
// translation layer the substrate was designed to make cheap:
// every MCP tool here is a one-line projection of an entry in
// Pulse's hand-authored capabilities manifest.
// Command pulse-mcp is the Model Context Protocol adapter for
// Pulse Intelligence. It exposes the same governed operations
// substrate that Pulse Assistant uses natively, but for
// stdio-speaking external agents like OpenCode, Claude Desktop,
// Claude Code, and other MCP clients. It is deliberately a thin
// translation layer: every MCP tool here is a one-line projection
// of an entry in Pulse's canonical capabilities manifest.
//
// Usage (typical Claude Desktop config entry):
// Usage (common mcpServers-style config entry):
//
// {
// "mcpServers": {
@ -24,39 +25,46 @@
//
// 1. Fetches /api/agent/capabilities from the configured Pulse
// instance at startup. The manifest is the single source of
// truth — adding a capability there automatically extends the
// MCP tool surface here, no MCP-side changes required.
// truth — adding a request/response capability to the Pulse MCP
// surface tool contract automatically extends the MCP tool
// surface here, no MCP-side changes required.
//
// 2. Translates each capability into an MCP tool with:
// - tool name = capability name (snake_case agent identifier)
// - description = capability description
// - inputSchema = derived from path placeholders + body shape:
// {resourceId} segments become required string properties;
// non-GET tools accept a free-form `body` object.
// - description = capability description plus manifest metadata
// (route, scope, action mode, approval policy, shapes, errors)
// - inputSchema = the manifest-owned JSON Schema when declared,
// with a path/body fallback for capabilities that have not yet
// authored a stricter schema.
//
// 3. Handles the MCP JSON-RPC methods Claude actually calls:
// initialize, tools/list, tools/call. Each tools/call
// 3. Handles the MCP JSON-RPC methods agents actually call:
// initialize, tools/list, tools/call, resources/list,
// resources/read, prompts/list, and prompts/get. Each tools/call
// resolves the manifest entry by name, substitutes path
// params, makes the HTTP request to Pulse with the configured
// token, and returns the JSON response (or stable error
// envelope) as a text content block.
//
// What it does not do (yet):
// What stays out of the request/response tool projection:
//
// - subscribe_events. SSE streaming doesn't fit the MCP tool
// shape; it would be an MCP "notification" or a long-running
// tool. Future slices can layer this on; agents that need
// real-time push consume the SSE stream directly.
// - subscribe_events as a callable tool. SSE streaming doesn't
// fit the MCP request/response shape, so --emit-notifications
// bridges it into JSON-RPC notifications when an agent needs
// real-time push.
//
// - Resource URIs. MCP supports `resources/list`/`resources/read`
// in addition to tools, but for Pulse the tool-only model is
// sufficient and keeps the adapter small. A future slice can
// project resource-context bundles as MCP resources if the
// UX value is clear.
// MCP resources are also manifest-backed: resources/list uses the
// canonical fleet context capability, and resources/read uses the
// canonical per-resource context capability. The resource URI shape
// and capability projection live in internal/agentcapabilities so
// this adapter does not grow a second inventory model.
//
// MCP prompts are manifest-backed workflow hints over that same
// tool/resource surface. The prompt catalog and prompt rendering live
// in internal/agentcapabilities so this adapter does not grow local
// operational playbooks.
package main
import (
"bufio"
"bytes"
"context"
"encoding/json"
@ -66,70 +74,26 @@ import (
"log"
"net/http"
"os"
"regexp"
"strings"
"sync"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/agentcapabilities"
)
// agentCapability mirrors Pulse's manifest wire shape — defined
// inline so the adapter depends on nothing in the pulse module.
type agentCapability struct {
Name string `json:"name"`
Description string `json:"description"`
Category string `json:"category"`
Method string `json:"method"`
Path string `json:"path"`
Scope string `json:"scope"`
ResponseShape string `json:"responseShape,omitempty"`
ErrorCodes []string `json:"errorCodes,omitempty"`
RequestBodyShape string `json:"requestBodyShape,omitempty"`
InputSchema json.RawMessage `json:"inputSchema,omitempty"`
}
type agentCapability = agentcapabilities.Capability
type agentCapabilitiesManifest = agentcapabilities.Manifest
type agentCapabilitiesManifest struct {
Version string `json:"version"`
Capabilities []agentCapability `json:"capabilities"`
}
type jsonRPCRequest = agentcapabilities.JSONRPCRequest
// jsonRPCRequest is the JSON-RPC 2.0 request envelope. Method is
// the MCP method name (e.g. "tools/list"); params is method-
// specific. ID is null for notifications, otherwise echoed back on
// the response.
type jsonRPCRequest struct {
JSONRPC string `json:"jsonrpc"`
ID json.RawMessage `json:"id,omitempty"`
Method string `json:"method"`
Params json.RawMessage `json:"params,omitempty"`
}
type jsonRPCResponse struct {
JSONRPC string `json:"jsonrpc"`
ID json.RawMessage `json:"id,omitempty"`
Result any `json:"result,omitempty"`
Error *jsonRPCError `json:"error,omitempty"`
}
type jsonRPCError struct {
Code int `json:"code"`
Message string `json:"message"`
Data any `json:"data,omitempty"`
}
// MCP tool shape. inputSchema is JSON Schema (draft-07-ish) that
// the agent uses to validate before calling.
type mcpTool struct {
Name string `json:"name"`
Description string `json:"description"`
InputSchema json.RawMessage `json:"inputSchema"`
}
// pathPlaceholderRE matches `{paramName}` in capability paths.
var pathPlaceholderRE = regexp.MustCompile(`\{([a-zA-Z][a-zA-Z0-9]*)\}`)
const (
pulseMCPServerName = "pulse-mcp"
pulseMCPServerVersion = "0.1.0"
)
func main() {
baseURL := flag.String("base-url", "http://localhost:7655", "Pulse base URL")
tokenEnv := flag.String("token-env", "PULSE_API_TOKEN", "Env var holding the Pulse API token")
baseURL := flag.String("base-url", agentcapabilities.DefaultMCPAdapterDefaultBaseURL, "Pulse base URL")
tokenEnv := flag.String("token-env", agentcapabilities.DefaultMCPAdapterTokenEnv, "Env var holding the Pulse API token")
emitNotifications := flag.Bool("emit-notifications", false, "Translate Pulse SSE events into JSON-RPC notifications on stdout. Off by default because not every MCP client surfaces server-initiated notifications; enable when wiring an autonomous agent that processes the JSON-RPC stream.")
flag.Parse()
@ -137,17 +101,18 @@ func main() {
log.SetPrefix("pulse-mcp ")
log.SetFlags(log.LstdFlags | log.Lshortfile)
token := strings.TrimSpace(os.Getenv(*tokenEnv))
if token == "" {
log.Fatalf("env var %s is empty; pulse-mcp needs an API token with monitoring:read scope (and monitoring:write for set/clear operator-state)", *tokenEnv)
}
manifest, err := fetchManifest(*baseURL)
manifestClient := &http.Client{Timeout: 10 * time.Second}
manifest, err := agentcapabilities.FetchManifest(context.Background(), manifestClient, *baseURL)
if err != nil {
log.Fatalf("could not fetch capabilities manifest from %s: %v", *baseURL, err)
}
log.Printf("fetched manifest %s with %d capabilities from %s", manifest.Version, len(manifest.Capabilities), *baseURL)
token := strings.TrimSpace(os.Getenv(*tokenEnv))
if token == "" {
log.Fatal(missingAPITokenMessage(*tokenEnv, manifest))
}
server := &mcpServer{
baseURL: *baseURL,
token: token,
@ -158,6 +123,14 @@ func main() {
server.serve(os.Stdin, os.Stdout)
}
func missingAPITokenMessage(tokenEnv string, manifest *agentCapabilitiesManifest) string {
scopeList := agentcapabilities.ManifestRequiredScopeList(manifest)
if scopeList == "" {
return fmt.Sprintf("env var %s is empty; pulse-mcp needs an API token for the manifest tools you enable", tokenEnv)
}
return fmt.Sprintf("env var %s is empty; pulse-mcp needs an API token. Mint one with the manifest scopes for the tools you enable (current manifest scopes: %s)", tokenEnv, scopeList)
}
// mcpServer holds the per-process state: the configured Pulse base
// URL and token, the manifest fetched at startup, and the HTTP
// client used to call Pulse.
@ -165,7 +138,7 @@ type mcpServer struct {
baseURL string
token string
manifest *agentCapabilitiesManifest
http *http.Client
http agentcapabilities.HTTPDoer
mu sync.Mutex // guards stdout writes
emitNotifications bool
// notificationsOnce ensures the SSE consumer goroutine starts
@ -180,49 +153,46 @@ type mcpServer struct {
out io.Writer
}
type pulseMCPAgentSurfaceHTTPDoer struct {
next agentcapabilities.HTTPDoer
}
func (d pulseMCPAgentSurfaceHTTPDoer) Do(req *http.Request) (*http.Response, error) {
if req != nil {
req = req.Clone(req.Context())
req.Header = req.Header.Clone()
req.Header.Set(agentcapabilities.AgentSurfaceHeader, agentcapabilities.AgentSurfacePulseMCP)
}
next := d.next
if next == nil {
next = http.DefaultClient
}
return next.Do(req)
}
// serve is the stdio loop: read line-delimited JSON-RPC requests
// from `in`, dispatch, write responses to `out`. Each request is on
// its own line; blank lines are ignored; EOF stops the server.
func (s *mcpServer) serve(in io.Reader, out io.Writer) {
s.out = out
scanner := bufio.NewScanner(in)
scanner.Buffer(make([]byte, 64*1024), 1<<22) // up to 4 MB per message
for scanner.Scan() {
line := strings.TrimSpace(scanner.Text())
if line == "" {
continue
}
var req jsonRPCRequest
if err := json.Unmarshal([]byte(line), &req); err != nil {
s.writeJSON(out, jsonRPCResponse{
JSONRPC: "2.0",
Error: &jsonRPCError{
Code: -32700, // Parse error
Message: fmt.Sprintf("malformed JSON-RPC request: %v", err),
},
})
continue
}
resp := s.dispatch(context.Background(), &req)
// Notifications (id is null/absent) get no response.
if len(req.ID) == 0 || string(req.ID) == "null" {
continue
}
s.writeJSON(out, resp)
}
if err := scanner.Err(); err != nil && err != io.EOF {
log.Printf("stdio scanner: %v", err)
if err := agentcapabilities.ServeJSONRPCLines(
context.Background(),
in,
func(ctx context.Context, req agentcapabilities.JSONRPCRequest) agentcapabilities.JSONRPCResponse {
return s.dispatch(ctx, &req)
},
func(resp agentcapabilities.JSONRPCResponse) error {
return s.writeJSON(out, resp)
},
); err != nil {
log.Printf("stdio JSON-RPC: %v", err)
}
}
// dispatch routes one JSON-RPC request to the right handler. The
// MCP methods we support are minimal: initialize, tools/list,
// tools/call. Anything else gets method-not-found per JSON-RPC.
func (s *mcpServer) dispatch(ctx context.Context, req *jsonRPCRequest) jsonRPCResponse {
resp := jsonRPCResponse{JSONRPC: "2.0", ID: req.ID}
switch req.Method {
case "initialize":
resp.Result = s.handleInitialize()
// dispatch delegates MCP method semantics to the shared Pulse Intelligence
// boundary. This binary owns session policy and stdout serialization only.
func (s *mcpServer) dispatch(ctx context.Context, req *jsonRPCRequest) agentcapabilities.JSONRPCResponse {
return agentcapabilities.DispatchMCPToolServerRequest(ctx, *req, s.manifestToolServer().Handlers(func() {
// Start the SSE-to-notifications bridge once per process,
// only if the operator opted in. Spawned after the
// initialize response is queued so the client sees the
@ -232,265 +202,79 @@ func (s *mcpServer) dispatch(ctx context.Context, req *jsonRPCRequest) jsonRPCRe
go s.consumeSSEEvents(context.Background())
})
}
case "tools/list":
resp.Result = s.handleToolsList()
case "tools/call":
result, err := s.handleToolsCall(ctx, req.Params)
if err != nil {
resp.Error = &jsonRPCError{Code: -32603, Message: err.Error()}
} else {
resp.Result = result
}
case "ping":
resp.Result = map[string]any{}
default:
resp.Error = &jsonRPCError{
Code: -32601,
Message: fmt.Sprintf("method not found: %s", req.Method),
}
}
return resp
}))
}
// handleInitialize returns the MCP server's capabilities. Tools
// are the only request/response category we expose. When the
// operator opts in via --emit-notifications, the server also
// publishes server-initiated notifications translated from
// Pulse's SSE event stream; resources, prompts, and sampling
// remain intentionally unadvertised.
func (s *mcpServer) handleInitialize() any {
caps := map[string]any{
"tools": map[string]any{},
func (s *mcpServer) manifestToolServer() agentcapabilities.MCPManifestToolServer {
var manifest agentcapabilities.Manifest
if s.manifest != nil {
manifest = *s.manifest
}
// MCP advertises notifications via experimental/extension
// keys today. We use a clearly-namespaced "experimental"
// block so clients that don't understand it ignore it
// silently rather than erroring on unknown keys.
if s.emitNotifications {
caps["experimental"] = map[string]any{
"pulseNotifications": map[string]any{
"kinds": []string{
"finding.created",
"approval.pending",
"action.completed",
},
},
}
}
return map[string]any{
"protocolVersion": "2024-11-05",
"capabilities": caps,
"serverInfo": map[string]any{
"name": "pulse-mcp",
"version": "0.1.0",
},
return agentcapabilities.MCPManifestToolServer{
ServerName: pulseMCPServerName,
ServerVersion: pulseMCPServerVersion,
SurfaceID: agentcapabilities.SurfaceIDPulseMCP,
EmitPulseNotifications: s.emitNotifications,
Client: s.pulseMCPHTTPClient(),
BaseURL: s.baseURL,
Token: s.token,
Manifest: manifest,
RecordWorkflowPromptActivity: s.recordWorkflowPromptActivity,
}
}
// handleToolsList projects each manifest capability into an MCP
// tool. subscribe_events is filtered out — SSE streaming doesn't
// fit the request/response tool shape.
func (s *mcpServer) handleToolsList() any {
tools := make([]mcpTool, 0, len(s.manifest.Capabilities))
for _, cap := range s.manifest.Capabilities {
if cap.Name == "subscribe_events" {
continue
}
schema := buildInputSchema(cap)
tools = append(tools, mcpTool{
Name: cap.Name,
Description: cap.Description,
InputSchema: schema,
})
func (s *mcpServer) pulseMCPHTTPClient() agentcapabilities.HTTPDoer {
if s == nil {
return pulseMCPAgentSurfaceHTTPDoer{}
}
return map[string]any{"tools": tools}
return pulseMCPAgentSurfaceHTTPDoer{next: s.http}
}
// handleToolsCall executes one tool invocation. params is shaped
// `{"name": "...", "arguments": {...}}`. The tool name resolves
// to a manifest capability; arguments fill path placeholders and
// (for non-GET tools) the request body.
func (s *mcpServer) handleToolsCall(ctx context.Context, raw json.RawMessage) (any, error) {
var params struct {
Name string `json:"name"`
Arguments map[string]any `json:"arguments"`
}
if err := json.Unmarshal(raw, &params); err != nil {
return nil, fmt.Errorf("decode tools/call params: %w", err)
}
cap, ok := s.findCapability(params.Name)
if !ok {
return nil, fmt.Errorf("unknown tool: %s", params.Name)
func (s *mcpServer) recordWorkflowPromptActivity(ctx context.Context, promptName string) {
promptName = strings.TrimSpace(promptName)
if s == nil || promptName == "" || strings.TrimSpace(s.baseURL) == "" || strings.TrimSpace(s.token) == "" {
return
}
ctx, cancel := context.WithTimeout(ctx, 2*time.Second)
defer cancel()
url, err := substitutePathParams(cap.Path, params.Arguments)
body, err := json.Marshal(map[string]string{"name": promptName})
if err != nil {
return nil, fmt.Errorf("substitute path params: %w", err)
log.Printf("workflow prompt activity: marshal: %v", err)
return
}
var body io.Reader
if cap.Method != http.MethodGet && cap.Method != http.MethodDelete {
// Non-GET/DELETE tools accept a `body` argument that's
// JSON-encoded as the request body. If absent, we send an
// empty object — that's fine for the finding-action
// capabilities that just need `{ "finding_id": "..." }`.
bodyArg, ok := params.Arguments["body"]
if !ok {
// Some capabilities take their body fields at the top
// level (no nested "body" key). Treat the whole
// arguments object as the body, minus any consumed
// path-placeholder keys.
pathParams := pathParamSet(cap.Path)
cleaned := map[string]any{}
for k, v := range params.Arguments {
if !pathParams[k] {
cleaned[k] = v
}
}
bodyArg = cleaned
}
buf, err := json.Marshal(bodyArg)
if err != nil {
return nil, fmt.Errorf("marshal request body: %w", err)
}
body = bytes.NewReader(buf)
}
httpReq, err := http.NewRequestWithContext(ctx, cap.Method, s.baseURL+url, body)
req, err := agentcapabilities.NewAgentHTTPRequest(
ctx,
http.MethodPost,
s.baseURL,
agentcapabilities.AgentWorkflowPromptActivityPath,
s.token,
bytes.NewReader(body),
)
if err != nil {
return nil, fmt.Errorf("build request: %w", err)
log.Printf("workflow prompt activity: %v", err)
return
}
httpReq.Header.Set("X-API-Token", s.token)
if body != nil {
httpReq.Header.Set("Content-Type", "application/json")
}
resp, err := s.http.Do(httpReq)
resp, err := s.pulseMCPHTTPClient().Do(req)
if err != nil {
return nil, fmt.Errorf("call Pulse: %w", err)
log.Printf("workflow prompt activity: %v", err)
return
}
defer resp.Body.Close()
respBody, err := io.ReadAll(resp.Body)
if err != nil {
return nil, fmt.Errorf("read response: %w", err)
_, _ = io.Copy(io.Discard, resp.Body)
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
log.Printf("workflow prompt activity: status %d", resp.StatusCode)
}
// Build the MCP content result. The substrate's stable error
// envelope ({"error": "code", "message": "..."}) is preserved
// verbatim — agents on the MCP side branch on the same code
// they would branching on the HTTP response.
text := string(respBody)
isError := resp.StatusCode < 200 || resp.StatusCode >= 300
return map[string]any{
"content": []map[string]any{
{
"type": "text",
"text": text,
},
},
"isError": isError,
}, nil
}
func (s *mcpServer) findCapability(name string) (agentCapability, bool) {
for _, c := range s.manifest.Capabilities {
if c.Name == name {
return c, true
}
}
return agentCapability{}, false
}
func (s *mcpServer) writeJSON(out io.Writer, v any) {
func (s *mcpServer) writeJSON(out io.Writer, v any) error {
s.mu.Lock()
defer s.mu.Unlock()
enc := json.NewEncoder(out)
enc.SetEscapeHTML(false)
if err := enc.Encode(v); err != nil {
if err := agentcapabilities.WriteJSONRPCMessage(out, v); err != nil {
log.Printf("encode response: %v", err)
return err
}
}
// buildInputSchema generates a permissive JSON Schema for a
// capability. Path placeholders become required string properties;
// non-GET/DELETE capabilities also accept a free-form `body`
// object. The schema is permissive on purpose — the manifest is
// the source of truth for what the underlying endpoint accepts;
// MCP just needs enough shape so the agent knows which fields to
// pass.
func buildInputSchema(cap agentCapability) json.RawMessage {
if len(cap.InputSchema) > 0 {
return cap.InputSchema
}
properties := map[string]any{}
required := []string{}
for _, m := range pathPlaceholderRE.FindAllStringSubmatch(cap.Path, -1) {
name := m[1]
properties[name] = map[string]any{
"type": "string",
"description": "Canonical " + name + " (e.g. \"vm:101\", \"container:web-1\")",
}
required = append(required, name)
}
if cap.Method != http.MethodGet && cap.Method != http.MethodDelete {
desc := "Request body fields"
if cap.RequestBodyShape != "" {
desc = "Request body fields. Shape hint: " + cap.RequestBodyShape
}
properties["body"] = map[string]any{
"type": "object",
"description": desc,
"additionalProperties": true,
}
}
schema := map[string]any{
"type": "object",
"properties": properties,
"additionalProperties": true,
}
if len(required) > 0 {
schema["required"] = required
}
out, _ := json.Marshal(schema)
return out
}
// substitutePathParams replaces `{name}` segments in a capability's
// path with the corresponding argument value. Missing args for
// declared placeholders are an error so the agent gets a stable
// failure rather than an HTTP 404 on a malformed URL.
func substitutePathParams(path string, args map[string]any) (string, error) {
var missing []string
out := pathPlaceholderRE.ReplaceAllStringFunc(path, func(match string) string {
name := match[1 : len(match)-1] // strip { and }
v, ok := args[name]
if !ok {
missing = append(missing, name)
return match
}
s, ok := v.(string)
if !ok {
missing = append(missing, name+" (not a string)")
return match
}
return s
})
if len(missing) > 0 {
return "", fmt.Errorf("missing path argument(s): %s", strings.Join(missing, ", "))
}
return out, nil
}
// pathParamSet returns the set of placeholder names declared in a
// path. Used to filter path args out of the request body when a
// caller passes everything at the top level.
func pathParamSet(path string) map[string]bool {
set := map[string]bool{}
for _, m := range pathPlaceholderRE.FindAllStringSubmatch(path, -1) {
set[m[1]] = true
}
return set
return nil
}
// consumeSSEEvents opens a long-lived SSE connection to Pulse's
@ -507,10 +291,9 @@ func pathParamSet(path string) map[string]bool {
// stays up across the kinds of stream stalls the MCP server's
// idle-tolerance budget already accepts on the substrate side.
func (s *mcpServer) consumeSSEEvents(ctx context.Context) {
url := s.baseURL + "/api/agent/events"
backoff := time.Second
for ctx.Err() == nil {
if err := s.streamSSEOnce(ctx, url); err != nil {
if err := s.streamSSEOnce(ctx, agentcapabilities.AgentEventsPath); err != nil {
if ctx.Err() != nil {
return
}
@ -532,95 +315,13 @@ func (s *mcpServer) consumeSSEEvents(ctx context.Context) {
}
// streamSSEOnce opens one SSE connection and reads events until
// the connection drops or the context is cancelled. Each parsed
// event becomes a JSON-RPC notification; the initial
// stream.connected event is skipped (it's a synchronization marker
// from the server, not an agent-actionable event).
func (s *mcpServer) streamSSEOnce(ctx context.Context, url string) error {
req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
if err != nil {
return err
}
req.Header.Set("X-API-Token", s.token)
req.Header.Set("Accept", "text/event-stream")
// Bypass the s.http client's Timeout — the SSE stream is
// long-lived; ctx is the right cancellation signal.
resp, err := http.DefaultClient.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return fmt.Errorf("subscribe: status %d", resp.StatusCode)
}
scanner := bufio.NewScanner(resp.Body)
scanner.Buffer(make([]byte, 64*1024), 1<<22)
var event, data string
for scanner.Scan() {
line := scanner.Text()
switch {
case strings.HasPrefix(line, "event: "):
event = strings.TrimPrefix(line, "event: ")
case strings.HasPrefix(line, "data: "):
data = strings.TrimPrefix(line, "data: ")
case line == "":
s.maybeEmitNotification(event, data)
event, data = "", ""
}
}
if err := scanner.Err(); err != nil {
return err
}
return nil
}
// maybeEmitNotification turns a parsed SSE record into a JSON-RPC
// notification, filtering out keepalives and the connect marker
// that aren't agent-actionable. The notification's params is the
// raw JSON of the SSE data field, so the receiver gets the same
// payload the substrate publishes.
func (s *mcpServer) maybeEmitNotification(event, data string) {
if event == "" || data == "" {
return
}
// Skip events that are pure transport plumbing — agents
// branching on event kind don't care about these.
if event == "stream.connected" || event == "heartbeat" {
return
}
var params json.RawMessage = []byte(data)
notification := jsonRPCRequest{
JSONRPC: "2.0",
Method: "notifications/" + event,
Params: params,
}
s.writeJSON(s.out, notification)
}
// fetchManifest pulls the capabilities manifest from Pulse. This
// is the only call the adapter makes before its first tool
// invocation; the manifest is not cached or refreshed during the
// process lifetime — restart pulse-mcp to pick up new
// capabilities.
func fetchManifest(baseURL string) (*agentCapabilitiesManifest, error) {
req, err := http.NewRequest(http.MethodGet, baseURL+"/api/agent/capabilities", nil)
if err != nil {
return nil, err
}
client := &http.Client{Timeout: 10 * time.Second}
resp, err := client.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("manifest GET returned %d", resp.StatusCode)
}
var m agentCapabilitiesManifest
if err := json.NewDecoder(resp.Body).Decode(&m); err != nil {
return nil, fmt.Errorf("decode manifest: %w", err)
}
return &m, nil
// the connection drops or the context is cancelled. The shared
// Pulse Intelligence core owns SSE-to-MCP notification projection;
// the adapter only serializes projected notifications to stdout.
func (s *mcpServer) streamSSEOnce(ctx context.Context, path string) error {
// Bypass the finite s.http timeout: the SSE stream is long-lived and ctx is
// the right cancellation signal. The wrapper still marks the adapter surface.
return agentcapabilities.StreamMCPEventNotifications(ctx, pulseMCPAgentSurfaceHTTPDoer{}, s.baseURL, path, s.token, func(notification agentcapabilities.JSONRPCRequest) error {
return s.writeJSON(s.out, notification)
})
}

1038
cmd/pulse-mcp/main_test.go
View file

File diff suppressed because it is too large Load diff