pulse-mcp-stream

pulse-mcp-stream

Layer 1: Real-Time Social Stream Monitoring via MCP

Version: 1.1.0 (music-topos enhanced) Trit: +1 (Generator - produces live data) Bundle: acquisition

Overview

Pulse-MCP-stream provides real-time monitoring of social interactions, enabling the cognitive surrogate system to stay updated with the latest patterns. It streams mentions, engagement changes, and trending topics.

Enhanced Integration: MCP + DuckDB

MCP Server (TypeScript)

// pulse-mcp-server/src/index.ts
import { Server } from "@modelcontextprotocol/sdk/server";
import { Firehose } from "@atproto/sync";
import * as duckdb from "duckdb";

const server = new Server({
  name: "pulse-mcp-stream",
  version: "1.0.0"
});

// Connect to DuckDB for persistence
const db = new duckdb.Database("pulse_stream.duckdb");

server.setRequestHandler("subscribe", async (params) => {
  const { actor, filters } = params;
  
  const firehose = new Firehose({
    service: "wss://bsky.network/xrpc/com.atproto.sync.subscribeRepos"
  });
  
  firehose.on("create", async (event) => {
    if (event.author === actor) {
      // Store in DuckDB
      await db.run(`
        INSERT INTO pulse_events (event_id, event_type, actor_did, text, created_at)
        VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP)
      `, [event.uri, event.type, event.author, event.record?.text]);
    }
  });
  
  await firehose.start();
  return { status: "subscribed", actor };
});

DuckDB Schema

CREATE TABLE pulse_events (
    event_id VARCHAR PRIMARY KEY,
    event_type VARCHAR,  -- 'post', 'reply', 'like', 'repost', 'mention'
    actor_did VARCHAR,
    actor_handle VARCHAR,
    subject_uri VARCHAR,
    text TEXT,
    created_at TIMESTAMP,
    ingested_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    gay_color VARCHAR  -- Deterministic color via SPI seed
);

CREATE TABLE engagement_deltas (
    delta_id VARCHAR PRIMARY KEY,
    post_uri VARCHAR,
    likes_delta INT,
    reposts_delta INT,
    replies_delta INT,
    velocity FLOAT,  -- engagements per minute
    measured_at TIMESTAMP
);

-- Real-time velocity tracking
CREATE VIEW v_post_velocity AS
SELECT 
    post_uri,
    COUNT(*) FILTER (WHERE event_type = 'like') as likes,
    COUNT(*) FILTER (WHERE event_type = 'repost') as reposts,
    COUNT(*) / (EXTRACT(EPOCH FROM (MAX(created_at) - MIN(created_at))) / 60.0) as velocity_per_min
FROM pulse_events
WHERE created_at > NOW() - INTERVAL '1 hour'
GROUP BY post_uri;