Integration Guide

🔌

DeerFlow Harness can be embedded into any Python application. This guide covers the integration patterns for using DeerFlow as a library inside your own system.

DeerFlow Harness is not only a standalone application. It is a Python library you can import and use inside your own backend, API server, automation system, or multi-agent orchestrator.

Embedding DeerFlowClient

The primary integration point is DeerFlowClient. It wraps the LangGraph runtime and exposes a clean API for sending messages and streaming responses from any Python application.


from deerflow.client import DeerFlowClient
 
# The client loads configuration on construction (reads config.yaml
# or DEER_FLOW_CONFIG_PATH)
client = DeerFlowClient()

The client is thread-safe and designed to be instantiated once and reused across requests.

Streaming

The recommended integration pattern is streaming. stream() is a sync generator, so no asyncio is required — it gives you real-time access to each token and event as the agent produces it:

Tool artifacts can contain native Python objects. default=str represents objects that are not JSON serializable as strings so these events do not interrupt the stream.


import json
 
def run_agent(thread_id: str, user_message: str):
    for event in client.stream(
        message=user_message,
        thread_id=thread_id,
        model_name="gpt-4o",
        subagent_enabled=True,
    ):
        # `stream()` yields `StreamEvent` dataclasses, not strings. Serialise each
        # one before it goes over the wire: Starlette calls `.encode()` on any
        # chunk that is not already `str` / `bytes`, so yielding the dataclass
        # raises `AttributeError` on the very first event.
        yield f"event: {event.type}\ndata: {json.dumps(event.data, default=str)}\n\n"
 
# In a FastAPI handler:
# from fastapi.responses import StreamingResponse
# return StreamingResponse(run_agent(thread_id, message), media_type="text/event-stream")

Non-streaming invocation

For batch processing or when you only need the final result:


def run_agent_sync(thread_id: str, user_message: str) -> str:
    result = client.chat(
        message=user_message,
        thread_id=thread_id,
    )
    return result

Thread management

Threads represent persistent conversations. Use unique thread IDs to isolate different user sessions:


import uuid
 
# New conversation
thread_id = str(uuid.uuid4())
 
# Continuing an existing conversation (same thread_id)
# The agent will see the full history if a checkpointer is configured
client.chat(message="Follow up question", thread_id=existing_thread_id)

Custom per-agent configuration

Build domain-specific agents by creating named agent configs and passing the agent_name when constructing the client:


# agents/research-assistant/config.yaml must exist with skills and tool config
 
research_client = DeerFlowClient(agent_name="research-assistant")
 
result = research_client.chat(
    message=user_message,
    thread_id=thread_id,
    model_name="gpt-4o",
)

Integrating with FastAPI

DeerFlow Gateway is itself a FastAPI application. You can mount it as a sub-application or router:


from fastapi import FastAPI
 
app = FastAPI()
 
# Mount the DeerFlow gateway router
from deerflow.app.gateway.main import app as gateway_app
app.mount("/deerflow", gateway_app)

Or use DeerFlowClient directly in your own FastAPI routes with streaming:


import json
 
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from deerflow.client import DeerFlowClient
 
app = FastAPI()
client = DeerFlowClient()
 
@app.post("/chat/{thread_id}")
async def chat(thread_id: str, body: dict):
    def generate():
        for event in client.stream(message=body["message"], thread_id=thread_id):
            # Serialise the `StreamEvent` dataclass; a raw f-string of the object
            # would ship its `repr()`, which no SSE client can parse.
            yield f"event: {event.type}\ndata: {json.dumps(event.data, default=str)}\n\n"
    return StreamingResponse(generate(), media_type="text/event-stream")

Integrating with LangGraph

DeerFlow Harness is built on LangGraph. The Lead Agent is a standard LangGraph graph. You can compose it with your own LangGraph nodes and graphs:


from deerflow.agents.lead_agent.agent import make_lead_agent
from langgraph.graph import StateGraph
 
# Access the underlying LangGraph agent factory
agent = make_lead_agent(config)

Configuration in embedded mode

When embedded in another application, set the config path explicitly to avoid ambiguity:


import os
os.environ["DEER_FLOW_CONFIG_PATH"] = "/path/to/my-deerflow-config.yaml"
 
from deerflow.client import DeerFlowClient
client = DeerFlowClient()

Set DEER_FLOW_CONFIG_PATH before constructing the client and keep it set for subsequent calls, since configuration is resolved again when the client runs.

MCP server integration

DeerFlow can expose its agent as an MCP server, allowing other MCP-compatible systems to call it as a tool. Refer to the DeerFlow repository for MCP server integration examples.

Customization Configuration