Lesson 14: Cross-Framework Integration

Lesson Goal

By the end of this lesson, you should be able to:

• Explain why wrapping external agents in MeshFlow governance is valuable.
• Use from_callable, from_langgraph, from_crewai, and from_autogen adapters.
• Wrap an external HTTP JSON service as a MeshFlow node.
• Mix adapter types in a single governed workflow.
• Understand what governance a wrapper adds and what it cannot change.

Estimated time: 45 to 60 minutes.

1. The Adapter Pattern

You have an existing agent. It might be a LangGraph compiled graph, a CrewAI crew, an AutoGen ConversableAgent, or a plain Python function. You want to run it inside a MeshFlow workflow so it gets:

• Cost and token tracking
• Carbon footprint measurement
• Guardian safety screening
• Ledger recording
• Circuit breaker protection
• Policy enforcement (budget, timeout, compliance)

The adapter pattern lets you add all of this without rewriting the agent. You wrap the existing object in a CallableAgent that MeshFlow can call.

Your existing agent → from_langgraph(agent) → CallableAgent → MeshFlow node

2. from_callable: The Universal Adapter

Any Python async function with the signature (task: str, context: dict) -> str can be wrapped:

from meshflow.agents.adapters import from_callable
from meshflow.core.schemas import AgentRole

async def my_research_fn(task: str, context: dict) -> str:
    return f"Research result for: {task}"

agent = from_callable(my_research_fn, role=AgentRole.RESEARCHER)

from_callable is the fallback adapter. Use it for:

• Functions you wrote yourself
• Third-party functions that match the signature
• Mock or test stubs during development

3. from_langgraph: Wrapping A LangGraph Runnable

LangGraph compiles graphs to a Runnable interface with an .invoke() method:

from meshflow.agents.adapters import from_langgraph
from meshflow.core.schemas import AgentRole

# compiled_graph = StateGraph(...).compile()
agent = from_langgraph(compiled_graph, role=AgentRole.RESEARCHER)

Under the hood, from_langgraph calls compiled_graph.invoke({"input": task}) and extracts the output. The LangGraph graph runs exactly as it was designed; MeshFlow wraps its execution with governance.

What MeshFlow adds:

• Cost and token tracking around the LangGraph invocation
• Timeout enforcement via step_timeout_s
• Circuit breaker protection
• Ledger recording of input and output

What it cannot change:

• How the LangGraph graph internally routes or retries
• The LangGraph graph's own state management
• Any tools the LangGraph nodes call directly

4. from_crewai: Wrapping A CrewAI Agent

from meshflow.agents.adapters import from_crewai
from meshflow.core.schemas import AgentRole

# crewai_agent = Agent(role="researcher", goal="...", backstory="...")
agent = from_crewai(crewai_agent, role=AgentRole.RESEARCHER)

The adapter calls crewai_agent.execute_task(task) and returns the result. The CrewAI agent's role, goal, and backstory remain unchanged.

5. from_autogen: Wrapping An AutoGen Agent

from meshflow.agents.adapters import from_autogen
from meshflow.core.schemas import AgentRole

# autogen_agent = AssistantAgent(name="researcher", ...)
agent = from_autogen(autogen_agent, role=AgentRole.RESEARCHER)

The adapter initiates a single-turn conversation with the AutoGen agent and returns the last message content. For multi-turn AutoGen setups, wrap the entire team as a single callable.

6. MeshNode.from_http: Wrapping An External Service

Any service that accepts a POST request with a JSON body and returns JSON can be wrapped as a MeshFlow node:

from meshflow.core.node import MeshNode

node = MeshNode.from_http(
    url="https://api.example.com/analyze",
    role=AgentRole.EXECUTOR,
)

The adapter sends {"task": task, "context": context} and reads response["output"] or response["result"] from the JSON response.

Use from_http for:

• Internal microservices you want to govern from MeshFlow
• Third-party APIs that accept task-like requests
• Staged rollouts where a new service takes traffic alongside an existing one

7. Mixing Adapter Types In One Workflow

You can combine all adapter types in a single workflow:

agents = [
    from_langgraph(lg_graph,    role=AgentRole.RESEARCHER),
    from_callable(my_analyzer,  role=AgentRole.EXECUTOR),
    from_crewai(crew_writer,    role=AgentRole.EXECUTOR),
    from_autogen(ag_reviewer,   role=AgentRole.CRITIC),
]

result = await Mesh(agents=agents, policy=policy).run(task)

Each agent runs in sequence. The output of one becomes the input task for the next. All agents are governed by the same policy, regardless of which framework produced them.

8. Progressive Adoption

The adapter pattern enables progressive migration. You do not have to rewrite your existing agents to get governance. You can:

1. Wrap all existing agents with from_callable or the appropriate adapter.
2. Run them under a MeshFlow policy to get cost visibility and circuit breakers.
3. Gradually replace wrapped agents with native MeshFlow nodes as you redesign.

This means you can start getting governance benefits on day one without a full rewrite.

9. What Adapters Cannot Fix

An adapter adds governance around an agent. It cannot change what happens inside:

• If a LangGraph graph has a bug, the adapter will record the bug but not fix it.
• If a CrewAI agent calls a dangerous tool, the adapter cannot intercept it

unless you also add guardian screening to the policy.

• Adapters do not validate agent outputs — they pass them through as-is. Use

explicit output schema validation in downstream nodes if you need that.

10. Hands-On Lab

MESHFLOW_MOCK=1 python3 hands_on/14_cross_framework.py

Observe:

• How each adapter wraps its agent with identical MeshFlow-side output
• The trace showing cost and token tracking for each wrapped agent
• The HITL gate between adapted agents (the human approval point)
• How from_http is demonstrated with a mock service

11. Summary

The adapter pattern — from_callable, from_langgraph, from_crewai, from_autogen, from_http — lets you wrap any existing agent in MeshFlow governance without rewriting it. All adapters produce CallableAgent objects that plug into Mesh(agents=[...]) or WorkflowDefinition nodes. Mixing adapter types in one workflow is fully supported. Governance is added around the agent; behavior inside the agent is unchanged.

Exercises

Exercise 1: Run the Cross-Framework Script and Identify Each Adapter

Goal: Observe multiple adapters working together in a single MeshFlow pipeline.

Instructions:

1. Run the hands-on script:

   python hands_on/14_cross_framework.py

1. Read through the output and the script source code. Identify each MeshNode and its source adapter type:

- Which nodes use from_callable? - Which nodes use from_langgraph, from_crewai, or from_autogen? - Which nodes (if any) use from_http?

1. For each node, record:

- The adapter type. - The node's role in the pipeline (what does it do?). - Whether the node appears in the MeshFlow trace output with a node_type label.

1. Answer: Does MeshFlow treat the nodes differently in the trace output based on their adapter type, or does it produce a uniform trace regardless of origin?

Expected output: A console trace showing each node's execution, with labels or metadata indicating the adapter type. All nodes should appear in a single unified execution trace.

---

Exercise 2: Wrap a Plain Python Function with from_callable

Goal: Practice the simplest adapter pattern.

Instructions:

1. Write a Python function that simulates a simple NLP task — for example, a keyword extractor:

   def extract_keywords(text: str) -> dict:
       words = text.lower().split()
       stopwords = {"the", "a", "an", "is", "in", "of", "and", "to"}
       keywords = [w for w in words if w not in stopwords and len(w) > 3]
       return {"keywords": list(set(keywords)), "count": len(set(keywords))}

1. Wrap it as a MeshNode:

   from meshflow import MeshNode
   node = MeshNode.from_callable(extract_keywords, name="keyword_extractor")

1. Add it to a minimal MeshFlow workflow and run it with a sample text input.
2. Inspect the trace output:

- Is keyword_extractor visible as a named step in the trace? - What does MeshFlow record as the step's input and output? - Is there a node_type: callable or similar label in the step metadata?

1. Now add a second from_callable node that takes the output of keyword_extractor and counts how many keywords start with a vowel. Chain the two nodes and run the workflow.

Expected output: A two-step trace showing both callable nodes with their respective inputs, outputs, and metadata. The workflow should complete without errors.

---

Exercise 3: Wrap a LangGraph Graph and Run It Inside MeshFlow

Goal: Integrate an existing LangGraph graph as a single MeshFlow node.

Instructions:

1. Create a minimal LangGraph graph (if you have LangGraph installed):

   from langgraph.graph import StateGraph, END

   def classify(state):
       text = state["text"]
       label = "positive" if "good" in text.lower() else "negative"
       return {"label": label}

   builder = StateGraph(dict)
   builder.add_node("classifier", classify)
   builder.set_entry_point("classifier")
   builder.add_edge("classifier", END)
   lg_graph = builder.compile()

1. Wrap it as a MeshNode:

   from meshflow import MeshNode
   node = MeshNode.from_langgraph(lg_graph, name="sentiment_graph")

1. Add it to a MeshFlow workflow alongside a plain from_callable node that prepares the input.
2. Run the workflow and inspect the trace. Note:

- Does the LangGraph graph's internal nodes appear as sub-steps in the MeshFlow trace, or does the entire graph appear as a single step? - What does node_type show for this step?

1. If LangGraph is not installed, read the hands-on script's LangGraph demo section and answer the inspection questions based on the printed output.

Expected output: The LangGraph graph appears as a single MeshFlow step with node_type: langgraph. The internal LangGraph state transitions are not individually visible in the MeshFlow trace (they are encapsulated by the adapter).

---

Exercise 4: Build a Mixed-Adapter Pipeline

Goal: Combine at least three different adapter types in one workflow.

Instructions:

1. Design a three-node pipeline that uses three different adapter types. For example:

- Node 1 (from_callable): A Python function that fetches data from a local file or generates mock data. - Node 2 (from_langgraph or from_crewai): A framework-specific component that processes the data. - Node 3 (from_http): An HTTP endpoint that a local test server or public API handles.

1. For the HTTP node, you can use a public API like https://httpbin.org/post or spin up a minimal local Flask server:

   # In a separate terminal:
   # pip install flask && python -c "
   # from flask import Flask, request, jsonify
   # app = Flask(__name__)
   # @app.route('/summarize', methods=['POST'])
   # def summarize():
   #     data = request.json
   #     return jsonify({'summary': data.get('text', '')[:50] + '...'})
   # app.run(port=8765)"
   node3 = MeshNode.from_http("http://localhost:8765/summarize", name="http_summarizer")

1. Wire the three nodes into a MeshFlow pipeline and run it.
2. Examine the final trace. Confirm that all three node types appear with distinct node_type labels and that data flows correctly from node 1 through to node 3.

Expected output: A three-step trace with different node_type values for each step, and the output of each step correctly passed as the input to the next.

---

Exercise 5: Progressive Adoption — Add MeshFlow to an Existing Script

Goal: Experience the progressive adoption pattern by instrumenting an existing codebase without rewriting it.

Instructions:

1. Take any existing Python script that calls one or more functions in sequence (it can be as simple as a data transformation pipeline with three functions).
2. Without rewriting any of the existing functions, wrap each one as a from_callable MeshNode:

   from meshflow import MeshFlow, MeshNode

   # Existing functions — unchanged
   def step_one(data): ...
   def step_two(data): ...
   def step_three(data): ...

   # Wrap without modifying
   app = MeshFlow()
   app.add_node(MeshNode.from_callable(step_one, name="step_one"))
   app.add_node(MeshNode.from_callable(step_two, name="step_two"))
   app.add_node(MeshNode.from_callable(step_three, name="step_three"))

   result = app.run({"input": "your data here"})

1. Run the instrumented version and compare the output to the original script's output — they should be identical.
2. Now enable the ledger and compliance modes:

   app = MeshFlow(enable_ledger=True, compliance=PolicyMode.STANDARD)

1. Observe that you have added full observability and compliance to the existing codebase with zero changes to the business logic functions.
2. Write two to three sentences reflecting on what this pattern means for teams with large existing codebases that want to adopt MeshFlow gradually.

Expected output: Identical business output to the original script, plus a full MeshFlow trace and ledger record with no changes to the wrapped functions.