Documentation

CrewAI — push and pull as native tools

from crewai import Agent, Task, Crew
from contextrelay.integrations.crewai import (
    ContextRelayPushTool, ContextRelayPullTool,
)

push = ContextRelayPushTool(api_key=API_KEY, channel="research-pipeline")
pull = ContextRelayPullTool(api_key=API_KEY)

researcher = Agent(role="Researcher",
    goal="Find sources on quantum computing", tools=[push])
writer     = Agent(role="Writer",
    goal="Draft article from research",       tools=[pull, push])
checker    = Agent(role="Fact-checker",
    goal="Verify each claim",                  tools=[pull])

crew = Crew(agents=[researcher, writer, checker],
            tasks=[...])
crew.kickoff()

LangGraph — pub/sub fan-out across nodes

from contextrelay import ContextRelay
relay = ContextRelay(api_key=API_KEY)

def researcher_node(state):
    findings = run_llm("research the topic")
    url = relay.push(findings, channel="research")
    return {"research_url": url}

def writer_node(state):
    findings = relay.pull(state["research_url"])
    return {"draft": run_llm(f"Write from: {findings}")}

# Add as nodes in your StateGraph — payloads stay at the edge

AutoGen — function tools

from autogen_agentchat.agents import AssistantAgent
from contextrelay.integrations.autogen import get_autogen_tools

tools = get_autogen_tools(api_key=API_KEY)

researcher = AssistantAgent(
    name="researcher",
    tools=tools,
    system_message="Push findings to ContextRelay, share the URL.",
)

The orchestrator (push and continue)

import os
from contextrelay import ContextRelay

relay = ContextRelay(api_key=os.environ["CONTEXTRELAY_API_KEY"])

# Fire a long-running research task
task_url = relay.push(
    "Analyse the last 30 days of customer support tickets for sentiment trends",
    channel="bg-tasks",
    metadata={"task_id": "T-1042", "priority": "low"},
)
# Orchestrator returns immediately — does NOT block on the worker

The worker (subscribe and respond)

import threading
from contextrelay import ContextRelay

relay = ContextRelay(api_key=os.environ["CONTEXTRELAY_API_KEY"])

def on_task(url):
    task = relay.pull(url)
    meta = relay.peek(url)              # cheap — no payload download
    result = run_long_analysis(task)    # could take minutes
    relay.push(
        result,
        channel="bg-done",
        metadata={"task_id": meta["task_id"], "duration_s": 142},
    )

threading.Thread(
    target=relay.subscribe,
    args=("bg-tasks", on_task),
    daemon=True,
).start()

The result-collector (subscribe to done)

def on_done(url):
    meta = relay.peek(url)
    print(f"Task {meta['task_id']} done in {meta['duration_s']}s")
    result = relay.pull(url)
    notify_user(meta["task_id"], result)

relay.subscribe("bg-done", on_done)   # blocking — run in a thread

Local Claude Code: push a debugging artifact

# Inside Claude Code, with the contextrelay MCP server registered:
# (~/.claude/mcp.json)

# Claude can call push_context as a native tool:
push_context(
  data="<full stack trace + repro steps>",
  channel="prod-issues",
  metadata={"severity": "high", "service": "api-gateway"},
)
# Returns: https://api.contextrelay.dev/pull/<uuid>

Cloud Worker: subscribe and triage

// Cloudflare Worker — listens on prod-issues
import { ContextRelay } from "contextrelay-js";

export default {
  async fetch(req, env) {
    const relay = new ContextRelay({ apiKey: env.CR_KEY });
    return relay.subscribe("prod-issues", async (url) => {
      const meta = await relay.peek(url);
      if (meta.severity === "high") {
        const trace = await relay.pull(url);
        await pageOnCall(meta.service, trace);
      }
    });
  },
};

What is encrypted vs what is not

Step 1 — paste this in Claude Code

Claude has the push_context MCP tool available. It will design the API and push the architecture to ContextRelay automatically.

You are a senior software architect. Design a production-ready FastAPI task management API.

Your design must cover:
- Data models: User, Task (with status, priority, due_date)
- All REST endpoints: auth (register/login/me), tasks (CRUD + filter by status)
- JWT authentication flow
- SQLite + SQLAlchemy ORM setup
- Pydantic schemas for request/response validation
- File structure and key implementation decisions

Write the complete architecture document with precise details so an engineer
can implement without asking questions.

When done, use the push_context tool to save the full document to ContextRelay.
Print the returned URL clearly — your engineer (Mistral) will build the entire codebase from it.

Step 2 — copy the URL Claude prints, paste this in Mistral

Replace PASTE_URL_FROM_CLAUDE_HERE and your API key. Mistral fetches the architecture and implements the full codebase.

You are a senior Python engineer. Your architect (Claude Opus) has designed a FastAPI API.

Step 1 — fetch the architecture from ContextRelay:

import requests
plan = requests.get(
    "PASTE_URL_FROM_CLAUDE_HERE",
    headers={"Authorization": "Bearer YOUR_API_KEY"}
).text
print(plan)

Step 2 — read the architecture and implement the complete codebase:
- Every Python file described in the design
- requirements.txt
- README.md with setup and run instructions

Rules:
- Match the architect's design exactly — do not improvise
- Output complete, runnable code only
- No placeholders, no TODOs

Automate the full loop in Python

import os, anthropic
from mistralai import Mistral
from contextrelay import ContextRelay

relay   = ContextRelay(api_key=os.environ["CONTEXTRELAY_API_KEY"])
claude  = anthropic.Anthropic()
mistral = Mistral(api_key=os.environ["MISTRAL_API_KEY"])

# ── Claude Opus: architect ───────────────────────────────────────
arch_response = claude.messages.create(
    model="claude-opus-4-5",
    max_tokens=8192,
    messages=[{
        "role": "user",
        "content": (
            "Design a production FastAPI task management API. Include data models, "
            "all endpoints, JWT auth, SQLAlchemy setup, and file structure. "
            "Be complete — an engineer will implement directly from this document."
        )
    }],
).content[0].text

# Push architecture — hand off just the URL
arch_url = relay.push(arch_response, metadata={"role": "architecture", "project": "task-api"})

# ── Mistral Large: engineer ──────────────────────────────────────
architecture = relay.pull(arch_url)   # Mistral fetches directly — 0 tokens in Claude

code_response = mistral.chat.complete(
    model="mistral-large-latest",
    messages=[{
        "role": "user",
        "content": (
            f"You are a senior Python engineer. Implement this architecture as a complete, "
            f"runnable codebase. Every file. No placeholders.\n\nArchitecture:\n{architecture}"
        )
    }],
).choices[0].message.content

# Push implementation — share the URL with your team or CI
impl_url = relay.push(code_response, metadata={"role": "implementation", "project": "task-api"})

Start the coordinator once (in your tmux session)

# terminal in your tmux session named "vibe", window 0
pip install contextrelay
contextrelay-bridge start --tmux vibe --task-channel vibe-tasks --done-channel vibe-done

Send tasks from any script

import os
from contextrelay import ContextRelay, AgentBridge

relay  = ContextRelay(api_key=os.environ["CONTEXTRELAY_API_KEY"])
bridge = AgentBridge(relay, task_channel="vibe-tasks", done_channel="vibe-done")

result = bridge.push_and_wait(
    "Refactor the auth module to use Firebase. "
    "Run the type checker. Return a summary of all changed files."
)

print(result)   # full Claude Code output, stripped of UI chrome