OpenAgentFlow MCP Tool Binding Architecture suggestion

readme.md

OpenAgentFlow: LangGraph + MCP Tool Binding Architecture

Our compiler translates human-readable .oaf topologies deterministically into executable LangGraph Python code.

We are currently implementing native support for tools and local OS tool execution and want to ensure our generated Python code aligns with LangGraph's recommended best practices.

The Architectural Goal

To keep the .oaf specification strictly platform-agnostic, we do not want users writing custom local tools.py sidecar files.

Instead, we are leveraging the MCPs. Developers declare their local MCP server in the .oaf file, and the OpenAgentFlow compiler generates the LangGraph execution boilerplate to handle the async MCP lifecycle, and inject them into the graph.

Specific Areas for Review

We would love a quick sanity check from the LangChain/LangGraph core team on the following:

1. Lifecycle Management: Is wrapping the StateGraph compilation and invocation inside the stdio_client() async context manager the safest way to maintain the MCP session throughout the lifecycle of the graph?
2. Dynamic Binding: We are using load_mcp_tools(session) to dynamically pull the schemas and pass them into ChatOpenAI(model="gpt-4o").bind_tools(mcp_tools) and the ToolNode. Is this the canonical pattern LangGraph envisions for zero-boilerplate MCP integration?

Context

• Project Repo: OpenAgentFlow/OpenAgentFlow
• Author: AbdulRahman Elzahaby

File 1: document_analyzer.oaf

// document_analyzer.oaf
// Developers declare the MCP server and required tools. 
// No local Python execution logic is required from the user.

workflow "Local Document Analyzer" {
  config {
    // Compiler maps this to the StdioServerParameters
    mcp_servers: ["npx -y @modelcontextprotocol/server-filesystem ./workspace"]
  }

  agent Reader {
    instructions: "Act as ..."
    model: "openai/gpt-4o"
    // Grants the agent access to dynamically loaded MCP tools
    tools: [mcp("filesystem:*")]
  }
  
  flow {
    start -> Reader -> end
  }
}

After prase and IR, This will be convert it to Genrated Python code File 2: compiled_langgraph_mcp_adapter.py

# auto-generated by OpenAgentFlow Compiler v0.1.5
import asyncio
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI

# Standard LangChain MCP integration libraries
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from langchain_mcp.tools import load_mcp_tools

class WorkflowState(TypedDict):
    messages: Annotated[list, add_messages]

async def execute_workflow():
    """
    Auto-generated LangGraph execution wrapper handling the async MCP server lifecycle.
    """
    
    # 1. Compiler securely injects MCP Server Configurations from the .oaf config block
    server_params = StdioServerParameters( 
        command="npx",
        args=["-y", "@modelcontextprotocol/server-filesystem", "./workspace"]
    )

    # 2. Manage the MCP context manager lifecycle
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # 3. Dynamically load tools via MCP and generate LangChain StructuredTools
            # This completely eliminates the need for a local tools.py file
            mcp_tools = await load_mcp_tools(session)
            
            # 4. Standard LangGraph Architecture
            tool_node = ToolNode(mcp_tools)
            llm = ChatOpenAI(model="gpt-4o").bind_tools(mcp_tools)

            async def agent_node(state: WorkflowState):
                response = await llm.ainvoke(state["messages"])
                return {"messages": [response]}

            # 5. Graph Assembly
            builder = StateGraph(WorkflowState)
            builder.add_node("Reader", agent_node)
            builder.add_node("tools", tool_node)
            
            builder.add_edge(START, "Reader") // auto-injected
            builder.add_conditional_edges("Reader", tools_condition)
            builder.add_edge("tools", "Reader")
            
            graph = builder.compile()
            
            # Execute the compiled graph...
            # await graph.ainvoke({"messages": [...]})

if __name__ == "__main__":
    asyncio.run(execute_workflow())

Thank you for your time and any feedback you can provide!