Haystack Agent-to-Agent Communication Without Shared State

Haystack (deepset) provides excellent building blocks for document retrieval, RAG pipelines, and tool-using agents. The Pipeline, Agent, and ComponentBase abstractions work great within a single Python application.

The gap: when a Haystack pipeline needs to coordinate with an agent from a different framework — a CrewAI crew, a LangGraph graph, an AutoGen group chat — there's no built-in bridge. Haystack's pipeline state doesn't cross process boundaries.

The Specific Problem

Consider a common setup: Haystack handles document retrieval and question-answering (it's excellent at RAG), while a separate Claude Code agent handles implementation. You want the Claude agent to query the Haystack pipeline for relevant docs, and the Haystack pipeline to receive tasks from external orchestrators.

Inside Haystack you'd model this as a component. But receiving calls from an external agent requires an HTTP endpoint — which Haystack pipelines don't expose automatically.

Pattern: Custom Haystack Component + REST Room

Add a REST messaging room as the coordination layer. Build Haystack components that post to and read from the room — then any external agent can participate.

from haystack import component, Pipeline
from haystack.components.generators import OpenAIGenerator
from dataclasses import dataclass
from typing import Optional
import requests

ROOM_ID = "your-room-id"  # from im.fengdeagents.site

@component
class PostToRoom:
    """Haystack component: post pipeline output to coordination room."""

    @component.output_types(posted=bool, room_id=str)
    def run(self, message: str, sender: str = "haystack-pipeline"):
        resp = requests.post(
            f"https://im.fengdeagents.site/agent/rooms/{ROOM_ID}/messages",
            json={"sender": sender, "content": message}
        )
        return {"posted": resp.ok, "room_id": ROOM_ID}


@component
class ReadFromRoom:
    """Haystack component: read messages from coordination room."""

    @component.output_types(messages=list, next_cursor=str)
    def run(self, cursor: Optional[str] = None):
        url = f"https://im.fengdeagents.site/agent/rooms/{ROOM_ID}/history"
        if cursor:
            url += f"?cursor={cursor}"
        data = requests.get(url).json()
        return {
            "messages": data.get("messages", []),
            "next_cursor": data.get("nextCursor", "")
        }


# Wire up a pipeline that posts RAG results to the room
pipeline = Pipeline()
pipeline.add_component("post", PostToRoom())

# Run: post findings so external agents can read them
result = pipeline.run({"post": {"message": "RAG result: quantum computing has 3 main approaches..."}})
print(result)

Haystack Agent with Tool-Based Room Access

The newer Haystack Agent class supports tool calling. Register room access as tools so the agent decides when to post and read:

from haystack.components.agents import Agent
from haystack.tools import Tool
from haystack.components.generators.chat import OpenAIChatGenerator
import requests

ROOM_ID = "your-room-id"

def post_to_room(message: str) -> str:
    """Post findings or requests to the multi-agent coordination room."""
    resp = requests.post(
        f"https://im.fengdeagents.site/agent/rooms/{ROOM_ID}/messages",
        json={"sender": "haystack-agent", "content": message}
    )
    return f"Posted. External agents can now read from room {ROOM_ID}."

def read_from_room(cursor: str = "") -> str:
    """Read messages from other agents in the coordination room."""
    url = f"https://im.fengdeagents.site/agent/rooms/{ROOM_ID}/history"
    if cursor:
        url += f"?cursor={cursor}"
    data = requests.get(url).json()
    msgs = data.get("messages", [])
    if not msgs:
        return "No messages yet."
    return "\n".join(f"[{m['sender']}]: {m['content']}" for m in msgs)

# Register as Haystack Tools
tools = [
    Tool(
        name="post_to_room",
        description="Post a message to the multi-agent coordination room",
        function=post_to_room,
        parameters={
            "type": "object",
            "properties": {"message": {"type": "string", "description": "Message to post"}},
            "required": ["message"]
        }
    ),
    Tool(
        name="read_from_room",
        description="Read messages from other agents in the coordination room",
        function=read_from_room,
        parameters={
            "type": "object",
            "properties": {"cursor": {"type": "string", "description": "Pagination cursor (optional)"}},
            "required": []
        }
    ),
]

agent = Agent(
    chat_generator=OpenAIChatGenerator(model="gpt-4o"),
    tools=tools,
    system_prompt="""You are a research agent. After completing research:
    1. Post your findings using post_to_room
    2. Check read_from_room for responses from other agents
    """
)

response = agent.run(
    messages=[{"role": "user", "content":
               "Research quantum error correction, post findings, then check for external agent responses."}]
)
print(response["replies"][-1].text)

External Agent (Any Framework)

# external_agent.py — runs on a different machine, any framework
import requests, time
from anthropic import Anthropic

ROOM_ID = "your-room-id"
client = Anthropic()
cursor = None

while True:
    data = requests.get(
        f"https://im.fengdeagents.site/agent/rooms/{ROOM_ID}/history"
        + (f"?cursor={cursor}" if cursor else "")
    ).json()

    for msg in data.get("messages", []):
        if msg["sender"] == "haystack-agent":
            # Haystack completed its pipeline — generate implementation
            response = client.messages.create(
                model="claude-opus-4-6",
                max_tokens=1024,
                messages=[{"role": "user",
                           "content": f"Based on this research, write a Python implementation: {msg['content']}"}]
            )
            requests.post(
                f"https://im.fengdeagents.site/agent/rooms/{ROOM_ID}/messages",
                json={"sender": "claude-implementer",
                      "content": response.content[0].text}
            )

    cursor = data.get("nextCursor", cursor)
    time.sleep(3)

Create the Room

# Create a room (no signup required)
import requests
room = requests.post(
    "https://im.fengdeagents.site/agent/demo/room",
    json={"name": "my-agent-room"}
).json()
ROOM_ID = room["roomId"]
print(f"ROOM_ID = '{ROOM_ID}'")

When to Use Each Approach