Multi-Agent Framework Integration

LangGraph agents publish tasks to Janus mailboxes and pull results from downstream agents. The Janus node acts as a reliable handoff between graph steps:

from langgraph.graph import StateGraph
from janus_sdk import JanusClient

client = JanusClient("http://janus:8080", tenant_id="acme")

class AgentState(TypedDict):
    pr_url: str
    review_result: dict

def review_node(state: AgentState):
    client.publish_task({
        "source_agent": "langgraph-review",
        "target_type": "mailbox",
        "target_value": "review-mb",
        "envelope": {"type": "review", "payload": state}
    })
    return state

def collect_node(state: AgentState):
    result = client.pull_task("langgraph-collect-mb", "collector")
    return {"review_result": result.task.envelope.payload}

graph = StateGraph(AgentState)
graph.add_node(review_node)
graph.add_node(collect_node)
graph.add_edge("review_node", "collect_node")

LangGraph handles the orchestration graph; Janus ensures tasks survive agent restarts and provides governance between steps.

AutoGen agents use Janus as a message bus for async team collaboration:

from autogen import AssistantAgent, UserProxyAgent
from janus_sdk import JanusClient

client = JanusClient("http://janus:8080", tenant_id="acme")

class JanusAgent(AssistantAgent):
    """AutoGen agent that sends/receives messages via Janus."""

    def send(self, message, recipient, **kwargs):
        client.publish_task({
            "source_agent": self.name,
            "target_type": "mailbox",
            "target_value": f"{recipient}-mb",
            "envelope": {"type": "autogen_msg", "payload": message}
        })

    def receive(self, message, sender, **kwargs):
        # Messages arrive through durable mailboxes
        # Agent pulls them when ready
        result = client.pull_task(f"{self.name}-mb", self.name)
        return result.task.envelope.payload

# AutoGen team with Janus backbone
reviewer = JanusAgent(name="reviewer", llm_config=llm_config)
coder = JanusAgent(name="coder", llm_config=llm_config)

CrewAI agents use Janus for persistent task handoff between crew members:

from crewai import Agent, Task, Crew
from janus_sdk import JanusClient

client = JanusClient("http://janus:8080", tenant_id="acme")

class JanusCrewAgent(Agent):
    """CrewAI agent with Janus-backed task queue."""

    def execute_task(self, task, context=None):
        # Publish task to Janus for the next crew member
        client.publish_task({
            "source_agent": self.role,
            "target_type": "mailbox",
            "target_value": f"{task.agent.role}-mb",
            "envelope": {"type": "crewai_task", "payload": task.description}
        })
        # Pull result from own mailbox
        result = client.pull_task(f"{self.role}-mb", self.role)
        return result.task.envelope.payload

# Crew with Janus resilience
crew = Crew(
    agents=[reviewer, coder, tester],
    tasks=[review_task, code_task, test_task],
    process=Process.sequential
)

GitHub Actions triggers Janus workflows and waits for agent results:

# .github/workflows/agent-review.yml
name: Agent Code Review
on: [pull_request]

jobs:
  review:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Request AI review via Janus
        run: |
          pip install janus-sdk
          python -c "
          from janus_sdk import JanusClient
          client = JanusClient('${{ secrets.JANUS_URL }}',
                              api_key='${{ secrets.JANUS_API_KEY }}')
          client.publish_task({
              'source_agent': 'github-actions',
              'target_type': 'mailbox',
              'target_value': 'review-mb',
              'envelope': {
                  'type': 'pr_review',
                  'payload': {'pr': '${{ github.event.pull_request.html_url }}'}
              }
          })
          "

      - name: Wait for review result
        run: |
          python -c "
          import time
          from janus_sdk import JanusClient
          client = JanusClient('${{ secrets.JANUS_URL }}',
                              api_key='${{ secrets.JANUS_API_KEY }}')
          while True:
              result = client.pull_task('github-results-mb', 'github-actions')
              if result:
                  print('Review complete:', result.task.envelope.payload)
                  break
              time.sleep(5)
          "