Real-Time Context for AI Agents

# One-node RisingWave for development. SQL port 4566, dashboard 5691.
docker run -it --pull=always -p 4566:4566 -p 5691:5691 \
  risingwavelabs/risingwave:latest single_node

# Connect with any PostgreSQL client.
psql -h localhost -p 4566 -d dev -U root

-- Append-only event streams from Kafka.
CREATE SOURCE orders (
  order_id    BIGINT,
  customer_id BIGINT,
  amount      NUMERIC,
  status      VARCHAR,          -- placed, paid, shipped, delivered, refunded
  order_ts    TIMESTAMPTZ,
  WATERMARK FOR order_ts AS order_ts - INTERVAL '30 seconds'
) WITH (
  connector = 'kafka',
  topic = 'orders',
  properties.bootstrap.server = 'broker:9092'
) FORMAT PLAIN ENCODE JSON;

CREATE SOURCE clickstream (
  customer_id BIGINT,
  event_type  VARCHAR,          -- page_view, add_to_cart, checkout_error
  page        VARCHAR,
  event_ts    TIMESTAMPTZ,
  WATERMARK FOR event_ts AS event_ts - INTERVAL '10 seconds'
) WITH (
  connector = 'kafka',
  topic = 'clickstream',
  properties.bootstrap.server = 'broker:9092'
) FORMAT PLAIN ENCODE JSON;

-- Help articles are published to a topic; every edit is a new version.
CREATE SOURCE help_articles (
  article_id  BIGINT,
  version     INT,
  title       VARCHAR,
  body        VARCHAR,
  published_at TIMESTAMPTZ
) WITH (
  connector = 'kafka',
  topic = 'help_articles',
  properties.bootstrap.server = 'broker:9092'
) FORMAT PLAIN ENCODE JSON;

-- Operational tables via CDC. Updates and deletes flow through.
CREATE SOURCE shop_pg WITH (
  connector = 'postgres-cdc',
  hostname = 'pg', port = '5432',
  username = 'rw', password = '...',
  database.name = 'shop', schema.name = 'public'
);

CREATE TABLE customers (
  customer_id BIGINT PRIMARY KEY,
  name        VARCHAR,
  tier        VARCHAR,
  region      VARCHAR
) FROM shop_pg TABLE 'public.customers';

CREATE TABLE tickets (
  ticket_id   BIGINT PRIMARY KEY,
  customer_id BIGINT,
  status      VARCHAR,          -- open, pending, resolved
  subject     VARCHAR,
  opened_at   TIMESTAMPTZ
) FROM shop_pg TABLE 'public.tickets';

-- Entity state: one row per customer, kept current incrementally.
CREATE MATERIALIZED VIEW customer_360 AS
SELECT c.customer_id, c.name, c.tier, c.region,
       COUNT(o.order_id) FILTER (WHERE o.status <> 'refunded') AS orders_total,
       COALESCE(SUM(o.amount) FILTER (WHERE o.status = 'paid'), 0) AS lifetime_value,
       COUNT(o.order_id) FILTER (WHERE o.status = 'refunded')     AS refunds_total,
       MAX(o.order_ts)                                            AS last_order_at,
       COUNT(t.ticket_id) FILTER (WHERE t.status <> 'resolved')   AS open_tickets
FROM customers c
LEFT JOIN orders  o ON o.customer_id = c.customer_id
LEFT JOIN tickets t ON t.customer_id = c.customer_id
GROUP BY c.customer_id, c.name, c.tier, c.region;

-- Recent activity: what this customer did in the last 15 minutes.
CREATE MATERIALIZED VIEW session_activity AS
SELECT customer_id,
       window_start, window_end,
       COUNT(*) FILTER (WHERE event_type = 'page_view')      AS page_views,
       COUNT(*) FILTER (WHERE event_type = 'add_to_cart')    AS cart_adds,
       COUNT(*) FILTER (WHERE event_type = 'checkout_error') AS checkout_errors,
       ARRAY_AGG(page ORDER BY event_ts DESC)                AS recent_pages
FROM HOP(clickstream, event_ts, INTERVAL '1 minute', INTERVAL '15 minutes')
GROUP BY customer_id, window_start, window_end;

-- Live status: the latest event per open order.
CREATE MATERIALIZED VIEW order_status_live AS
SELECT order_id, customer_id, amount, status, order_ts AS updated_at
FROM (
  SELECT *, ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY order_ts DESC) AS rn
  FROM orders
) t
WHERE rn = 1 AND status NOT IN ('delivered', 'refunded');

-- Retrievable knowledge: embed each article version as it is published.
-- The function is defined once so the model and key live in one place.
CREATE FUNCTION embed(VARCHAR) RETURNS VECTOR(1536) LANGUAGE SQL AS $$
  SELECT openai_embedding(
    '{"model": "text-embedding-3-small", "api_key": "sk-..."}'::jsonb, $1
  )::vector(1536)
$$;

CREATE MATERIALIZED VIEW article_embeddings AS
SELECT article_id, version, title, body, published_at,
       embed(title || E'\n' || body) AS embedding
FROM help_articles;

-- Vector indexes are built on append-only inputs; this view is one.
CREATE INDEX article_embeddings_hnsw ON article_embeddings
USING HNSW (embedding)
INCLUDE (article_id, version, title, body)
WITH (distance_type = 'cosine', m = 16, ef_construction = 64);

-- Agent memory: the agent writes here, views read it back.
CREATE TABLE agent_memory (
  memory_id    BIGINT PRIMARY KEY,
  customer_id  BIGINT,
  kind         VARCHAR,         -- promise, preference, escalation, summary
  content      VARCHAR,
  created_at   TIMESTAMPTZ
);

CREATE MATERIALIZED VIEW customer_memory AS
SELECT customer_id,
       ARRAY_AGG(kind || ': ' || content ORDER BY created_at DESC) AS notes
FROM agent_memory
WHERE created_at > NOW() - INTERVAL '90 days'
GROUP BY customer_id;

-- The whole structured context for one customer, from one snapshot.
SELECT to_jsonb(c)                         AS profile,
       to_jsonb(s)                         AS last_15_min,
       (SELECT jsonb_agg(to_jsonb(o))
          FROM order_status_live o
         WHERE o.customer_id = c.customer_id) AS open_orders,
       m.notes                             AS memory
FROM customer_360 c
LEFT JOIN session_activity s
       ON s.customer_id = c.customer_id
      AND s.window_end = (SELECT MAX(window_end) FROM session_activity
                           WHERE customer_id = c.customer_id)
LEFT JOIN customer_memory m ON m.customer_id = c.customer_id
WHERE c.customer_id = $1;

-- The three most relevant help articles for the customer's question,
-- latest version only, embedded with the same function as the index.
SELECT DISTINCT ON (article_id) article_id, title, body
FROM article_embeddings
ORDER BY article_id, version DESC, embedding <=> embed($1)
LIMIT 3;

import json, re, psycopg, anthropic

rw = psycopg.connect("postgresql://agent_ro@localhost:4566/dev", autocommit=True)
llm = anthropic.Anthropic()  # any LLM client works; the context is plain JSON

CONTEXT_SQL = open("context.sql").read()      # the first query above
RETRIEVE_SQL = open("retrieve.sql").read()    # the second query above

def bind(sql: str) -> str:
    """The SQL files use Postgres $n placeholders; psycopg wants %(name)s."""
    return re.sub(r"\$(\d+)", r"%(p\1)s", sql)

def build_context(customer_id: int, question: str) -> dict:
    with rw.cursor() as cur:
        cur.execute(bind(CONTEXT_SQL), {"p1": customer_id})
        profile, last_15_min, open_orders, memory = cur.fetchone()
        cur.execute(bind(RETRIEVE_SQL), {"p1": question})
        articles = [{"id": a, "title": t, "body": b} for a, t, b in cur.fetchall()]
    return {
        "profile": profile,
        "last_15_min": last_15_min,
        "open_orders": open_orders or [],
        "memory": memory or [],
        "relevant_articles": articles,
    }

def answer(customer_id: int, question: str) -> str:
    ctx = build_context(customer_id, question)
    resp = llm.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=600,
        system=(
            "You are a support agent. Use only the facts in <context>. "
            "The context is at most a few seconds old; treat it as current."
        ),
        messages=[{
            "role": "user",
            "content": f"<context>{json.dumps(ctx, default=str)}</context>\n\n{question}",
        }],
    )
    return resp.content[0].text

print(answer(42, "Where is my order and can I still change the address?"))

{
  "mcpServers": {
    "risingwave": {
      "command": "python",
      "args": ["/path/to/risingwave-mcp/src/main.py"],
      "env": {
        "MCP_TRANSPORT": "stdio",
        "RISINGWAVE_CONNECTION_STR": "postgresql://agent_ro@localhost:4566/dev"
      }
    }
  }
}