mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-07-09 16:00:59 +00:00
Translate Pulse SSE events into MCP notifications when opted in
Closes the documented limitation in slice 51's pulse-mcp: MCP clients that process server-initiated notifications can now react to Pulse's push channel without holding a separate HTTP connection to /api/agent/events. The bridge is opt-in via --emit-notifications because not every MCP client surfaces arbitrary notifications/* methods (Claude Desktop, today, does not). Autonomous agents that consume the JSON-RPC stream programmatically benefit; UI-mediated clients should keep the flag off and use the SSE stream directly. Implementation: a long-lived goroutine, started once after the first initialize, that opens /api/agent/events, parses the substrate's wire format, and emits a JSON-RPC notification per non-transport event. Method names mirror the SSE event kinds (notifications/finding.created, notifications/approval. pending, notifications/action.completed). Params is the SSE data payload verbatim so agents see the same wire shape an HTTP SSE consumer would. stream.connected and heartbeat are filtered as transport plumbing. The consumer reconnects with capped exponential backoff on transient errors. When --emit-notifications is on, initialize advertises the supported event kinds under capabilities.experimental.pulseNotifications.kinds. Clients that don't understand the experimental block ignore it silently. Three tests pin the behaviour: the initialize handshake's capability block is correctly gated on the flag; the notification filter rejects transport events and accepts the three substrate kinds; an httptest.NewServer-backed end-to-end translates a multi-event SSE stream into JSON-RPC notifications with the substrate's payload preserved. Also flagged in AGENT_SUBSTRATE.md "what it does not do yet": the action-execution endpoints (/api/actions/plan, decision, execute) emit a different error envelope from the agent surface (APIError with stable code under "code") versus the agent-stable shape (stable code under "error"). Adding them to the manifest requires resolving that mismatch first; recorded as a focused slice for whenever the substrate's reach extends to direct agent-driven execution.
This commit is contained in:
rcourtman
parent
f2d9d2aba8
commit
818221b457
5 changed files with 388 additions and 29 deletions
|
|
@ -78,10 +78,40 @@ Add the server to your project's `.mcp.json` (or your user-level config):
|
|||
| ---- | ------- | ------- |
|
||||
| `--base-url` | `http://localhost:7655` | Pulse instance to talk to |
|
||||
| `--token-env` | `PULSE_API_TOKEN` | Env var holding the API token |
|
||||
| `--emit-notifications` | `false` | Translate Pulse SSE events into JSON-RPC notifications on stdout |
|
||||
|
||||
The token is always read from an environment variable, never a flag, so
|
||||
it does not appear in process listings.
|
||||
|
||||
### About `--emit-notifications`
|
||||
|
||||
When enabled, the server opens a long-lived connection to
|
||||
`/api/agent/events` after `initialize` and writes a JSON-RPC
|
||||
notification on stdout for every non-transport event:
|
||||
|
||||
```json
|
||||
{ "jsonrpc": "2.0", "method": "notifications/finding.created", "params": { ... } }
|
||||
{ "jsonrpc": "2.0", "method": "notifications/approval.pending", "params": { ... } }
|
||||
{ "jsonrpc": "2.0", "method": "notifications/action.completed", "params": { ... } }
|
||||
```
|
||||
|
||||
The `params` object is the SSE event's `data` payload verbatim, so an
|
||||
agent that already knows the substrate's wire shape sees identical
|
||||
content to what an HTTP SSE consumer would. Transport plumbing
|
||||
(`stream.connected`, `heartbeat`) is filtered out.
|
||||
|
||||
It is off by default because not every MCP client surfaces
|
||||
server-initiated notifications. Enable it when wiring an autonomous
|
||||
agent that processes the JSON-RPC stream programmatically. Claude
|
||||
Desktop today does not surface arbitrary `notifications/*` methods
|
||||
in the chat UI; if your client falls in that category, leave the
|
||||
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.
|
||||
|
||||
## What the tools do
|
||||
|
||||
The exact set is whatever your Pulse instance's manifest declares. As of
|
||||
|
|
@ -140,12 +170,14 @@ call.
|
|||
|
||||
## Known limitations
|
||||
|
||||
- **No `subscribe_events`.** SSE streaming does not fit the MCP
|
||||
- **No `subscribe_events` tool.** SSE streaming does not fit the MCP
|
||||
request/response tool shape, so the adapter does not expose the
|
||||
`/api/agent/events` stream. Agents that need real-time push (finding
|
||||
created, approval pending, action completed) consume the SSE stream
|
||||
directly via HTTP. A future version may layer this onto MCP
|
||||
notifications.
|
||||
`/api/agent/events` stream as a callable tool. Agents that want
|
||||
real-time push have two options: consume the SSE stream directly
|
||||
via HTTP (works with any MCP client), or run with
|
||||
`--emit-notifications` so the bridge translates SSE events into
|
||||
JSON-RPC notifications on the stdio channel (requires a client
|
||||
that processes server-initiated notifications).
|
||||
|
||||
- **Manifest is fetched once.** The server fetches `/api/agent/
|
||||
capabilities` at startup and does not refresh during the process
|
||||
|
|
|
|||
|
|
@ -129,6 +129,7 @@ var pathPlaceholderRE = regexp.MustCompile(`\{([a-zA-Z][a-zA-Z0-9]*)\}`)
|
|||
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")
|
||||
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()
|
||||
|
||||
log.SetOutput(os.Stderr)
|
||||
|
|
@ -147,10 +148,11 @@ func main() {
|
|||
log.Printf("fetched manifest %s with %d capabilities from %s", manifest.Version, len(manifest.Capabilities), *baseURL)
|
||||
|
||||
server := &mcpServer{
|
||||
baseURL: *baseURL,
|
||||
token: token,
|
||||
manifest: manifest,
|
||||
http: &http.Client{Timeout: 30 * time.Second},
|
||||
baseURL: *baseURL,
|
||||
token: token,
|
||||
manifest: manifest,
|
||||
http: &http.Client{Timeout: 30 * time.Second},
|
||||
emitNotifications: *emitNotifications,
|
||||
}
|
||||
server.serve(os.Stdin, os.Stdout)
|
||||
}
|
||||
|
|
@ -159,17 +161,29 @@ func main() {
|
|||
// URL and token, the manifest fetched at startup, and the HTTP
|
||||
// client used to call Pulse.
|
||||
type mcpServer struct {
|
||||
baseURL string
|
||||
token string
|
||||
manifest *agentCapabilitiesManifest
|
||||
http *http.Client
|
||||
mu sync.Mutex // guards stdout writes
|
||||
baseURL string
|
||||
token string
|
||||
manifest *agentCapabilitiesManifest
|
||||
http *http.Client
|
||||
mu sync.Mutex // guards stdout writes
|
||||
emitNotifications bool
|
||||
// notificationsOnce ensures the SSE consumer goroutine starts
|
||||
// at most once per process — `initialize` may be called more
|
||||
// than once if a client reconnects, but we only need one
|
||||
// consumer per stdio session.
|
||||
notificationsOnce sync.Once
|
||||
// out is the writer used for both responses and notifications.
|
||||
// Captured the first time `serve` is called so the SSE consumer
|
||||
// can write notifications to the same channel without an extra
|
||||
// argument-passing dance.
|
||||
out io.Writer
|
||||
}
|
||||
|
||||
// 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() {
|
||||
|
|
@ -208,6 +222,15 @@ func (s *mcpServer) dispatch(ctx context.Context, req *jsonRPCRequest) jsonRPCRe
|
|||
switch req.Method {
|
||||
case "initialize":
|
||||
resp.Result = s.handleInitialize()
|
||||
// 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
|
||||
// handshake reply before any notification arrives.
|
||||
if s.emitNotifications {
|
||||
s.notificationsOnce.Do(func() {
|
||||
go s.consumeSSEEvents(context.Background())
|
||||
})
|
||||
}
|
||||
case "tools/list":
|
||||
resp.Result = s.handleToolsList()
|
||||
case "tools/call":
|
||||
|
|
@ -229,14 +252,33 @@ func (s *mcpServer) dispatch(ctx context.Context, req *jsonRPCRequest) jsonRPCRe
|
|||
}
|
||||
|
||||
// handleInitialize returns the MCP server's capabilities. Tools
|
||||
// are the only category we expose; resources, prompts, and
|
||||
// sampling are intentionally not advertised.
|
||||
// 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{},
|
||||
}
|
||||
// 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": map[string]any{
|
||||
"tools": map[string]any{},
|
||||
},
|
||||
"capabilities": caps,
|
||||
"serverInfo": map[string]any{
|
||||
"name": "pulse-mcp",
|
||||
"version": "0.1.0",
|
||||
|
|
@ -446,6 +488,112 @@ func pathParamSet(path string) map[string]bool {
|
|||
return set
|
||||
}
|
||||
|
||||
// consumeSSEEvents opens a long-lived SSE connection to Pulse's
|
||||
// /api/agent/events stream and translates each non-keepalive event
|
||||
// into a JSON-RPC notification on stdout. Notifications use the
|
||||
// MCP convention `notifications/<event-kind>` (e.g.
|
||||
// `notifications/finding.created`) so MCP clients that route on
|
||||
// method names can dispatch directly.
|
||||
//
|
||||
// The function is meant to run for the lifetime of the stdio
|
||||
// session; it returns only on context cancellation or unrecoverable
|
||||
// stream errors. On a recoverable error (network blip, server
|
||||
// restart), it backs off briefly and reconnects so the bridge
|
||||
// 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 ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
log.Printf("SSE bridge: %v (reconnecting in %s)", err, backoff)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-time.After(backoff):
|
||||
}
|
||||
if backoff < 30*time.Second {
|
||||
backoff *= 2
|
||||
}
|
||||
continue
|
||||
}
|
||||
// Clean exit (e.g. server closed the stream); reset the
|
||||
// backoff so the next reconnect is immediate.
|
||||
backoff = time.Second
|
||||
}
|
||||
}
|
||||
|
||||
// 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
|
||||
|
|
|
|||
|
|
@ -308,3 +308,163 @@ func TestServer_NotificationGetsNoResponse(t *testing.T) {
|
|||
t.Errorf("notification produced output; want silent. got: %s", out.String())
|
||||
}
|
||||
}
|
||||
|
||||
// TestServer_InitializeAdvertisesNotificationsCapabilityWhenEnabled
|
||||
// pins the discovery contract for the SSE bridge: when
|
||||
// --emit-notifications is on, the initialize response advertises
|
||||
// the kinds an MCP client can expect under
|
||||
// experimental.pulseNotifications.kinds. Drift in either
|
||||
// direction (advertising when disabled, or omitting when enabled)
|
||||
// breaks client expectations about whether to wait for pushes.
|
||||
func TestServer_InitializeAdvertisesNotificationsCapabilityWhenEnabled(t *testing.T) {
|
||||
t.Run("disabled by default", func(t *testing.T) {
|
||||
s := &mcpServer{manifest: &agentCapabilitiesManifest{}}
|
||||
result := s.handleInitialize().(map[string]any)
|
||||
caps := result["capabilities"].(map[string]any)
|
||||
if _, ok := caps["experimental"]; ok {
|
||||
t.Error("initialize must NOT advertise pulseNotifications when --emit-notifications is off")
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("advertised when enabled", func(t *testing.T) {
|
||||
s := &mcpServer{manifest: &agentCapabilitiesManifest{}, emitNotifications: true}
|
||||
result := s.handleInitialize().(map[string]any)
|
||||
caps := result["capabilities"].(map[string]any)
|
||||
exp, ok := caps["experimental"].(map[string]any)
|
||||
if !ok {
|
||||
t.Fatal("initialize must advertise experimental block when --emit-notifications is on")
|
||||
}
|
||||
pn, ok := exp["pulseNotifications"].(map[string]any)
|
||||
if !ok {
|
||||
t.Fatal("experimental block must contain pulseNotifications descriptor")
|
||||
}
|
||||
kinds, ok := pn["kinds"].([]string)
|
||||
if !ok || len(kinds) == 0 {
|
||||
t.Fatalf("pulseNotifications.kinds must list the SSE event kinds; got %v", pn["kinds"])
|
||||
}
|
||||
want := map[string]bool{"finding.created": false, "approval.pending": false, "action.completed": false}
|
||||
for _, k := range kinds {
|
||||
if _, exists := want[k]; exists {
|
||||
want[k] = true
|
||||
}
|
||||
}
|
||||
for k, seen := range want {
|
||||
if !seen {
|
||||
t.Errorf("pulseNotifications.kinds missing %q", k)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// TestServer_MaybeEmitNotificationFiltersTransportEvents pins that
|
||||
// the bridge skips events that are pure transport plumbing
|
||||
// (stream.connected, heartbeat). Forwarding those would surface
|
||||
// noise an agent has no useful action on, and would teach
|
||||
// downstream code to filter them client-side instead of trusting
|
||||
// the substrate's "doorbell, not transport" intent.
|
||||
func TestServer_MaybeEmitNotificationFiltersTransportEvents(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
event, data string
|
||||
shouldEmit bool
|
||||
methodPrefix string
|
||||
}{
|
||||
{"finding.created passes through", "finding.created", `{"findingId":"f1"}`, true, "notifications/finding.created"},
|
||||
{"approval.pending passes through", "approval.pending", `{"approvalId":"a1"}`, true, "notifications/approval.pending"},
|
||||
{"action.completed passes through", "action.completed", `{"actionId":"x1"}`, true, "notifications/action.completed"},
|
||||
{"stream.connected is filtered", "stream.connected", `{}`, false, ""},
|
||||
{"heartbeat is filtered", "heartbeat", "", false, ""},
|
||||
{"empty event is filtered", "", `{"x":1}`, false, ""},
|
||||
{"empty data is filtered", "finding.created", "", false, ""},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
out := &bytes.Buffer{}
|
||||
s := &mcpServer{out: out}
|
||||
s.maybeEmitNotification(tc.event, tc.data)
|
||||
if tc.shouldEmit {
|
||||
if out.Len() == 0 {
|
||||
t.Fatalf("expected notification on stdout for %q; got nothing", tc.event)
|
||||
}
|
||||
body := out.String()
|
||||
if !strings.Contains(body, `"method":"`+tc.methodPrefix+`"`) {
|
||||
t.Errorf("expected method %q; got %s", tc.methodPrefix, body)
|
||||
}
|
||||
if !strings.Contains(body, `"jsonrpc":"2.0"`) {
|
||||
t.Errorf("notification must be JSON-RPC 2.0; got %s", body)
|
||||
}
|
||||
// Notifications must NOT carry an id field per
|
||||
// JSON-RPC 2.0 spec — clients that see an id treat
|
||||
// the message as a request and may try to respond.
|
||||
if strings.Contains(body, `"id":`) {
|
||||
t.Errorf("notification must omit the id field; got %s", body)
|
||||
}
|
||||
} else {
|
||||
if out.Len() != 0 {
|
||||
t.Errorf("expected silence for %q event; got %s", tc.event, out.String())
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestServer_StreamSSEOnceTranslatesEventsToNotifications is the
|
||||
// integration test for the bridge: spin up a fake SSE server
|
||||
// emitting the substrate's wire format, point the consumer at it,
|
||||
// and assert each non-transport event lands as a JSON-RPC
|
||||
// notification on the configured out writer.
|
||||
func TestServer_StreamSSEOnceTranslatesEventsToNotifications(t *testing.T) {
|
||||
sseBody := strings.Join([]string{
|
||||
"event: stream.connected",
|
||||
"data: {}",
|
||||
"",
|
||||
"event: finding.created",
|
||||
"data: {\"findingId\":\"f1\",\"severity\":\"critical\"}",
|
||||
"",
|
||||
"event: heartbeat",
|
||||
"",
|
||||
"event: action.completed",
|
||||
"data: {\"actionId\":\"x1\",\"success\":true}",
|
||||
"",
|
||||
}, "\n") + "\n"
|
||||
|
||||
pulse := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Header.Get("X-API-Token") != "test-token" {
|
||||
w.WriteHeader(http.StatusUnauthorized)
|
||||
return
|
||||
}
|
||||
w.Header().Set("Content-Type", "text/event-stream")
|
||||
_, _ = w.Write([]byte(sseBody))
|
||||
}))
|
||||
defer pulse.Close()
|
||||
|
||||
out := &bytes.Buffer{}
|
||||
s := &mcpServer{
|
||||
baseURL: pulse.URL,
|
||||
token: "test-token",
|
||||
http: pulse.Client(),
|
||||
out: out,
|
||||
}
|
||||
if err := s.streamSSEOnce(context.Background(), pulse.URL+"/api/agent/events"); err != nil {
|
||||
t.Fatalf("streamSSEOnce: %v", err)
|
||||
}
|
||||
|
||||
body := out.String()
|
||||
if !strings.Contains(body, `"method":"notifications/finding.created"`) {
|
||||
t.Errorf("missing finding.created notification; got %s", body)
|
||||
}
|
||||
if !strings.Contains(body, `"method":"notifications/action.completed"`) {
|
||||
t.Errorf("missing action.completed notification; got %s", body)
|
||||
}
|
||||
if strings.Contains(body, "stream.connected") {
|
||||
t.Errorf("stream.connected must be filtered out as transport plumbing; got %s", body)
|
||||
}
|
||||
if strings.Contains(body, `"method":"notifications/heartbeat"`) {
|
||||
t.Errorf("heartbeat must be filtered out; got %s", body)
|
||||
}
|
||||
// The payload data must round-trip verbatim so agents see the
|
||||
// substrate's wire shape unchanged.
|
||||
if !strings.Contains(body, `"findingId":"f1"`) {
|
||||
t.Errorf("notification params must round-trip the SSE data field; got %s", body)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -61,20 +61,29 @@ manifest-driven:
|
|||
and Pulse's tools appear natively. The adapter projects each
|
||||
manifest capability into one MCP tool with auto-derived input
|
||||
schema; adding capabilities to Pulse extends the MCP surface
|
||||
without changes in the adapter.
|
||||
without changes in the adapter. Run with `--emit-notifications`
|
||||
to also translate Pulse's SSE events (`finding.created`,
|
||||
`approval.pending`, `action.completed`) into JSON-RPC
|
||||
notifications on the stdio channel so autonomous MCP-bound agents
|
||||
can react to push events without holding a separate HTTP
|
||||
connection.
|
||||
|
||||
## What it does not do yet
|
||||
|
||||
- The MCP adapter does not project the SSE event stream as MCP
|
||||
notifications. Agents that need real-time push consume the SSE
|
||||
stream directly via HTTP.
|
||||
|
||||
- The substrate does not yet expose the governed action-execution
|
||||
surface (`/api/actions/plan`, `/api/actions/{id}/execute`) as
|
||||
agent-stable manifest entries. Action governance flows through the
|
||||
existing approval store and the `action.completed` push channel;
|
||||
direct agent-driven execution is the natural next slice if you
|
||||
want to pick this back up.
|
||||
surface (`/api/actions/plan`, `/api/actions/{id}/decision`,
|
||||
`/api/actions/{id}/execute`) as agent-stable manifest entries.
|
||||
Those handlers exist and are wired through the action audit store,
|
||||
but they emit a different error envelope from the agent surface
|
||||
(`APIError` shape: stable code under `code`, human message under
|
||||
`error`) versus the agent-stable shape (stable code under `error`,
|
||||
human under `message`). Adding them to the manifest as-is would
|
||||
force agents to remember which envelope each capability uses.
|
||||
Resolving that mismatch is a focused slice of its own. Until then,
|
||||
action governance flows through the existing approval store and
|
||||
the `action.completed` push channel: the AI service plans, an
|
||||
operator (or operator-acting agent) approves, the substrate emits
|
||||
the event.
|
||||
|
||||
## Provable claims
|
||||
|
||||
|
|
|
|||
|
|
@ -1516,6 +1516,16 @@ over stdio with line-delimited framing, and preserves the
|
|||
substrate's stable error envelope (`{"error": "code", "message":
|
||||
"..."}`) verbatim through MCP's content-and-isError result so
|
||||
agents on the MCP side branch on the same stable codes.
|
||||
Optionally, with `--emit-notifications`, the adapter also
|
||||
subscribes to `/api/agent/events` and translates each
|
||||
non-transport SSE event into a JSON-RPC notification on stdout
|
||||
(`notifications/finding.created`, `notifications/approval.pending`,
|
||||
`notifications/action.completed`); the notification's `params` is
|
||||
the SSE `data` payload verbatim so MCP-bound autonomous agents
|
||||
react to pushes without holding a separate HTTP connection. The
|
||||
flag is off by default because not every MCP client surfaces
|
||||
server-initiated notifications; transport plumbing
|
||||
(`stream.connected`, `heartbeat`) is filtered.
|
||||
|
||||
The integration guide for `cmd/pulse-mcp` lives at
|
||||
`cmd/pulse-mcp/README.md`. It carries the canonical Claude Desktop
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue