MCP & Protocols2026-02-04•1,793 words•8 min read

MCP Workflow Chaining - Deep Dive

#mcp#vision

MCP Workflow Chaining - Deep Dive

Problem Statement

I have multiple MCP servers (vision-agent-mcp, code-mcp, prompts-engine), but no way to:

Chain tools across multiple servers

Pass outputs from one tool as inputs to another

Schedule and repeat workflows

Handle errors and retries

This is the missing piece for autonomous multi-agent orchestration.

MCP Protocol Capabilities

What MCP Provides

1. JSON-RPC 2.0 Protocol

Request:
  {
    "jsonrpc": "2.0",
    "id": 1,
    "method": "tools/call",
    "params": {"name": "navigate", "arguments": {"url": "..."}}
  }

Response:
  {
    "jsonrpc": "2.0",
    "id": 1,
    "result": {...}
  }

2. Three Communication Models

Method 1: Direct Tool Calls

# Call tool directly
response = call_mcp_method(
    server="vision-agent-mcp",
    method="tools/call",
    params={"name": "screenshot"}
)

Method 2: Resources (Data Passing)

# Read resource from one server
data = read_resource(
    server="vision-agent-mcp",
    uri="screenshots/latest"
)

# Write to another server
write_resource(
    server="code-mcp",
    uri="input/screenshot",
    data=data
)

Method 3: Prompts (Workflow Templates)

# Execute prompt with variables
result = execute_prompt(
    server="vision-agent-mcp",
    name="analyze_dashboard",
    arguments={"dashboard_url": "..."}
)

What MCP Does NOT Provide

No workflow chaining - Can't connect tools across servers

No variable passing - Outputs not automatically inputs

No scheduling - No built-in timers or repeats

No error handling - No retries or fallbacks

No orchestration - No control flow beyond prompts

Solution: MCP Workflow Engine

Architecture

┌──────────────────────────────────────────────────────┐
│              MCP Workflow Engine                    │
│                                                      │
│  ┌────────────────────────────────────────────┐    │
│  │           Workflow Definition              │    │
│  │  - Chain of steps (tool/resource/prompt)  │    │
│  │  - Variables (inputs, intermediate, outputs)│    │
│  │  - Scheduling (cron, interval, one-shot)   │    │
│  │  - Error handling (retries, fallbacks)    │    │
│  └────────────────────────────────────────────┘    │
│                      ↓                               │
│  ┌────────────────────────────────────────────┐    │
│  │           Workflow Executor                 │    │
│  │  - Parse workflow definition                │    │
│  │  - Execute steps in sequence                │    │
│  │  - Pass variables between steps            │    │
│  │  - Handle errors and retries               │    │
│  │  - Track progress and results               │    │
│  └────────────────────────────────────────────┘    │
│                      ↓                               │
│  ┌────────────────────────────────────────────┐    │
│  │         MCP Client Layer                   │    │
│  │  - Connect to multiple MCP servers         │    │
│  │  - Send JSON-RPC requests                 │    │
│  │  - Receive and parse responses             │    │
│  │  - Handle connection failures              │    │
│  └────────────────────────────────────────────┘    │
└──────────────────────────────────────────────────────┘
                     ↓
┌──────────────────────────────────────────────────────┐
│              MCP Servers                             │
│  ┌──────────────┐  ┌──────────────┐  ┌───────────┐  │
│  │ vision-agent │  │   code-mcp   │  │ prompts   │  │
│  │     -mcp     │  │              │  │ -engine   │  │
│  │              │  │              │  │           │  │
│  │ 9 tools      │  │ 3 tools      │  │ execute    │  │
│  │ 4 resources  │  │              │  │ prompts   │  │
│  │ 4 prompts    │  │              │  │           │  │
│  └──────────────┘  └──────────────┘  └───────────┘  │
└──────────────────────────────────────────────────────┘

Workflow Definition Schema

Simple Chain

{
  "name": "daily_site_check",
  "description": "Check a website and save results",
  "schedule": {
    "type": "cron",
    "expression": "0 9 * * *"
  },
  "variables": {
    "input": {
      "site_url": "https://example.com"
    },
    "output": {
      "screenshot_path": null,
      "analysis_result": null
    }
  },
  "steps": [
    {
      "id": "step1",
      "server": "vision-agent-mcp",
      "type": "tool",
      "action": "navigate",
      "arguments": {
        "url": "{{variables.input.site_url}}"
      }
    },
    {
      "id": "step2",
      "server": "vision-agent-mcp",
      "type": "tool",
      "action": "screenshot",
      "arguments": {},
      "save_output_to": "variables.output.screenshot_path"
    },
    {
      "id": "step3",
      "server": "vision-agent-mcp",
      "type": "resource",
      "action": "read",
      "uri": "screenshots/latest",
      "save_output_to": "variables.output.screenshot_data"
    },
    {
      "id": "step4",
      "server": "code-mcp",
      "type": "tool",
      "action": "execute",
      "arguments": {
        "code": "import json; print(json.dumps({'status': 'OK', 'timestamp': '2026-02-04'}))"
      },
      "save_output_to": "variables.output.analysis_result"
    }
  ]
}

Conditional Workflow

{
  "name": "check_and_alert",
  "steps": [
    {
      "id": "check",
      "server": "vision-agent-mcp",
      "type": "tool",
      "action": "verify",
      "arguments": {"text": "Error"},
      "on_error": "alert_step"
    },
    {
      "id": "alert_step",
      "server": "vision-agent-mcp",
      "type": "prompt",
      "action": "execute",
      "prompt_name": "send_alert",
      "arguments": {"message": "Error detected!"}
    }
  ],
  "conditionals": {
    "check_to_alert": {
      "condition": "{{step_check.output}} contains 'Error'",
      "if_true": ["alert_step"],
      "if_false": []
    }
  }
}

Parallel Execution

{
  "name": "multi_site_check",
  "steps": [
    {
      "id": "parallel_checks",
      "type": "parallel",
      "tasks": [
        {
          "id": "check_site1",
          "server": "vision-agent-mcp",
          "type": "tool",
          "action": "verify",
          "arguments": {"url": "https://site1.com"}
        },
        {
          "id": "check_site2",
          "server": "vision-agent-mcp",
          "type": "tool",
          "action": "verify",
          "arguments": {"url": "https://site2.com"}
        },
        {
          "id": "check_site3",
          "server": "vision-agent-mcp",
          "type": "tool",
          "action": "verify",
          "arguments": {"url": "https://site3.com"}
        }
      ]
    },
    {
      "id": "aggregate",
      "server": "code-mcp",
      "type": "tool",
      "action": "execute",
      "arguments": {
        "code": "import json; results = [check_site1.output, check_site2.output, check_site3.output]; print(json.dumps({'total': len(results), 'passed': sum(1 for r in results if 'OK' in r)}))"
      }
    }
  ]
}

Loop Until Condition

{
  "name": "wait_for_element",
  "steps": [
    {
      "id": "wait_loop",
      "type": "loop",
      "max_iterations": 60,
      "delay_seconds": 1,
      "until": "{{step_check.output}} contains 'Expected Text'",
      "steps": [
        {
          "id": "check",
          "server": "vision-agent-mcp",
          "type": "tool",
          "action": "screenshot",
          "arguments": {}
        },
        {
          "id": "extract",
          "server": "vision-agent-mcp",
          "type": "resource",
          "action": "read",
          "uri": "screenshots/latest"
        }
      ]
    }
  ]
}

Variable Passing

Types of Variables

Input Variables - Defined in workflow, provided at runtime

Step Outputs - Captured from step execution

Global Variables - Shared across all steps

Saved Variables - Persisted between workflow runs

Variable Substitution Syntax

{{variables.input.site_url}}        # Input variable
{{step1.output}}                     # Step output
{{variables.global.timestamp}}       # Global variable
{{saved.last_screenshot}}            # Saved from previous run

Variable Scope

variables.input                      # Read-only, set at start
variables.output                     # Write-only, final results
variables.global                     # Read-write, shared
variables.saved                      # Persisted between runs

Error Handling

Retry Strategies

{
  "id": "step1",
  "server": "vision-agent-mcp",
  "type": "tool",
  "action": "navigate",
  "arguments": {"url": "{{variables.input.url}}"},
  "retry": {
    "max_attempts": 3,
    "delay_seconds": 5,
    "backoff": "exponential",
    "on_fail": "fallback_step"
  }
}

Fallback Steps

{
  "id": "primary_step",
  "server": "vision-agent-mcp",
  "type": "tool",
  "action": "screenshot",
  "on_error": "fallback_step"
},
{
  "id": "fallback_step",
  "server": "vision-agent-mcp",
  "type": "tool",
  "action": "navigate",
  "arguments": {"url": "https://backup.example.com"}
}

Error Callbacks

{
  "on_error": {
    "step": "send_alert",
    "arguments": {"message": "Workflow failed at step {{failed_step.id}}: {{error.message}}"}
  }
}

Scheduling

Cron Schedules

{
  "schedule": {
    "type": "cron",
    "expression": "0 9 * * *",
    "timezone": "UTC"
  }
}

Interval Schedules

{
  "schedule": {
    "type": "interval",
    "seconds": 300
  }
}

One-Shot (Scheduled)

{
  "schedule": {
    "type": "once",
    "at": "2026-02-04T14:30:00Z"
  }
}

MCP Client Implementation

Connection Pool

class MCPServerConnection:
    """Maintains connection to a single MCP server."""

    def __init__(self, server_path: str):
        self.server_path = server_path
        self.process = None

    def connect(self):
        """Start the MCP server process."""
        self.process = subprocess.Popen(
            [self.server_path],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE
        )

    def call(self, method: str, params: dict) -> dict:
        """Send JSON-RPC request and return response."""
        request = {
            "jsonrpc": "2.0",
            "id": self._next_id(),
            "method": method,
            "params": params
        }
        self.process.stdin.write(json.dumps(request).encode())
        self.process.stdin.flush()

        response = json.loads(self.process.stdout.readline().decode())
        return response.get("result", {})

Multi-Server [REDACTED]

    """Manages connections to multiple MCP servers."""

    def __init__(self):
        self.connections = {}

    def add_server(self, name: str, server_path: str):
        """Add an MCP server."""
        conn = MCPServerConnection(server_path)
        conn.connect()
        self.connections[name] = conn

    def call_tool(self, server: str, tool_name: str, arguments: dict) -> dict:
        """Call a tool on a specific server."""
        if server not in self.connections:
            raise ValueError(f"Server not connected: {server}")

        return self.connections[server].call(
            method="tools/call",
            params={
                "name": tool_name,
                "arguments": arguments
            }
        )

    def read_resource(self, server: str, uri: str) -> dict:
        """Read a resource from a specific server."""
        return self.connections[server].call(
            method="resources/read",
            params={"uri": uri}
        )

    def execute_prompt(self, server: str, prompt_name: str, arguments: dict) -> dict:
        """Execute a prompt on a specific server."""
        return self.connections[server].call(
            method="prompts/execute",
            params={
                "name": prompt_name,
                "arguments": arguments
            }
        )

Workflow Execution

Sequential Execution

async def execute_workflow(workflow: dict, mcp_[REDACTED]: MCPServer[REDACTED]):
    """Execute a workflow step by step."""

    # Initialize variables
    variables = {
        "input": workflow.get("variables", {}).get("input", {}),
        "output": {},
        "global": {
            "start_time": datetime.now().isoformat()
        }
    }

    # Execute each step
    for step in workflow["steps"]:
        step_id = step["id"]

        try:
            # Substitute variables in arguments
            arguments = substitute_variables(step["arguments"], variables)

            # Execute based on type
            if step["type"] == "tool":
                result = mcp_[REDACTED].call_tool(
                    server=step["server"],
                    tool_name=step["action"],
                    arguments=arguments
                )
            elif step["type"] == "resource":
                result = mcp_[REDACTED].read_resource(
                    server=step["server"],
                    uri=step["uri"]
                )
            elif step["type"] == "prompt":
                result = mcp_[REDACTED].execute_prompt(
                    server=step["server"],
                    prompt_name=step["action"],
                    arguments=arguments
                )

            # Save output if specified
            if "save_output_to" in step:
                save_variable(variables, step["save_output_to"], result)

            variables[step_id] = {"output": result}

        except Exception as e:
            if "on_error" in step:
                # Handle error
                pass
            else:
                raise

    return variables["output"]

Variable Substitution

def substitute_variables(template: Any, variables: dict) -> Any:
    """Replace {{variable}} placeholders with actual values."""

    if isinstance(template, str):
        # Replace {{variables.input.url}} with actual value
        result = template
        for match in re.finditer(r'\{\{([^}]+)\}\}', template):
            var_path = match.group(1)
            value = get_nested_value(variables, var_path)
            result = result.replace(match.group(0), str(value))
        return result

    elif isinstance(template, dict):
        # Recursively substitute in dict values
        return {k: substitute_variables(v, variables) for k, v in template.items()}

    elif isinstance(template, list):
        # Recursively substitute in list items
        return [substitute_variables(item, variables) for item in template]

    else:
        # No substitution needed
        return template

def get_nested_value(data: dict, path: str) -> Any:
    """Get value from nested dict using dot notation."""
    keys = path.split(".")
    value = data
    for key in keys:
        if key in value:
            value = value[key]
        else:
            return None
    return value

Practical Examples

Example 1: Daily Website Check

{
  "name": "daily_site_check",
  "schedule": {"type": "cron", "expression": "0 9 * * *"},
  "variables": {
    "input": {"site_url": "https://mysite.com"}
  },
  "steps": [
    {"id": "navigate", "server": "vision-agent-mcp", "type": "tool", "action": "navigate", "arguments": {"url": "{{variables.input.site_url}}"}},
    {"id": "screenshot", "server": "vision-agent-mcp", "type": "tool", "action": "screenshot"},
    {"id": "analyze", "server": "vision-agent-mcp", "type": "prompt", "action": "execute", "prompt_name": "analyze_dashboard", "arguments": {"dashboard_url": "{{variables.input.site_url}}"}},
    {"id": "save", "server": "code-mcp", "type": "tool", "action": "execute", "arguments": {"code": "with open('/tmp/check_result.txt', 'w') as f: f.write(json.dumps({{'timestamp': '2026-02-04', 'status': 'OK'}}))"}}
  ]
}

Example 2: Multi-Server Form Test

{
  "name": "form_test_workflow",
  "variables": {
    "input": {"form_url": "https://example.com/form", "test_data": {"username": "test", "password": "test"}}
  },
  "steps": [
    {"id": "navigate", "server": "vision-agent-mcp", "type": "tool", "action": "navigate", "arguments": {"url": "{{variables.input.form_url}}"}},
    {"id": "fill", "server": "vision-agent-mcp", "type": "tool", "action": "fill_form", "arguments": {"url": "{{variables.input.form_url}}", "fields": "{{variables.input.test_data}}"}},
    {"id": "screenshot", "server": "vision-agent-mcp", "type": "tool", "action": "screenshot"},
    {"id": "verify", "server": "vision-agent-mcp", "type": "tool", "action": "verify", "arguments": {"text": "Success"}},
    {"id": "log", "server": "code-mcp", "type": "tool", "action": "execute", "arguments": {"code": "print('Form test passed')"}}
  ]
}

Example 3: Parallel Site Comparison

{
  "name": "compare_sites",
  "variables": {
    "input": {"sites": ["https://site1.com", "https://site2.com", "https://site3.com"]}
  },
  "steps": [
    {
      "id": "parallel_screenshots",
      "type": "parallel",
      "tasks": [
        {"id": "ss1", "server": "vision-agent-mcp", "type": "tool", "action": "navigate", "arguments": {"url": "{{variables.input.sites[0]}}"}},
        {"id": "ss2", "server": "vision-agent-mcp", "type": "tool", "action": "navigate", "arguments": {"url": "{{variables.input.sites[1]}}"}},
        {"id": "ss3", "server": "vision-agent-mcp", "type": "tool", "action": "navigate", "arguments": {"url": "{{variables.input.sites[2]}}"}}
      ]
    },
    {"id": "compare", "server": "vision-agent-mcp", "type": "prompt", "action": "execute", "prompt_name": "compare_multiple_sites", "arguments": {"sites": "{{variables.input.sites}}"}}
  ]
}