LangGraph workflows

For installation, typed tools, permissions, screenshots, and client lifecycle, start with LangChain and LangGraph. This page focuses on graph-specific orchestration choices.

Native BeeOS tools in a graph

langchain-beeos returns standard LangChain tools that can be passed directly to LangGraph’s ToolNode:

import os

from langgraph.prebuilt import ToolNode
from langchain_beeos import BeeOSToolkit


with BeeOSToolkit(
    instance_id=os.environ["BEEOS_INSTANCE_ID"],
    include_tools={"beeos_computer_info", "beeos_mobile_info"},
) as toolkit:
    beeos_node = ToolNode(toolkit.get_tools())
    # Build, compile, and invoke the graph while the toolkit remains open.

Use include_tools, fixed target IDs, and mutation scopes before compiling the graph. A checkpoint restores graph state; it must not silently broaden the BeeOS tools or permissions available to a resumed run.

Choose MCP or OpenAPI

Use MCP when BeeOS should be a tool available to a model-driven graph. Use OpenAPI when the graph itself owns task creation, retries, cancellation, and state transitions.

MCP tools in a graph

Install the LangChain MCP adapter:

pip install langgraph langchain-mcp-adapters

Load the BeeOS agent as a remote Streamable HTTP server:

import os

from langchain.agents import create_agent
from langchain_mcp_adapters.client import MultiServerMCPClient


async def build_agent():
    agent_id = os.environ["BEEOS_AGENT_ID"]
    client = MultiServerMCPClient(
        {
            "beeos": {
                "transport": "http",
                "url": f"https://mcp.beeos.ai/{agent_id}/mcp",
                "headers": {
                    "Authorization": f"Bearer {os.environ['BEEOS_API_KEY']}"
                },
            }
        }
    )
    tools = await client.get_tools()
    return create_agent(os.environ["LLM_MODEL"], tools)

Invoke it with an explicit policy in the graph state or system prompt. Do not let a generic research graph silently inherit permission to send messages, purchase items, or modify accounts on a real phone.

Durable OpenAPI node

For long-running work, create a BeeOS task from a regular async graph node and store taskId in graph state. A later node can poll or resume from a webhook. This makes LangGraph checkpointing and BeeOS task durability reinforce each other instead of holding one MCP request open.

import os
import httpx


async def create_beeos_task(state: dict) -> dict:
    headers = {"Authorization": f"Bearer {os.environ['BEEOS_API_KEY']}"}
    body = {
        "message": state["phone_task"],
        "idempotency_key": state["run_id"],
    }
    async with httpx.AsyncClient(base_url="https://openapi.beeos.ai") as client:
        response = await client.post(
            f"/api/v1/agents/{state['agent_id']}/tasks",
            headers=headers,
            json=body,
        )
        response.raise_for_status()
        return {"beeos_task": response.json()}

Confirm the current request schema in Calling Agents before copying this node into production; generated BeeOS SDKs can replace the raw HTTP call. See the official LangChain MCP adapter documentation and Choosing a Protocol.

Getting Started

Guides

Architecture

A2A Protocol

MCP

Integrations

SDKs

Reference

LangGraph workflows

Native BeeOS tools in a graph

Choose MCP or OpenAPI

MCP tools in a graph

Durable OpenAPI node

​Native BeeOS tools in a graph

​Choose MCP or OpenAPI

​MCP tools in a graph

​Durable OpenAPI node

Native BeeOS tools in a graph

Choose MCP or OpenAPI

MCP tools in a graph

Durable OpenAPI node