#241 Highway MCP Server - Enable LLMs to Generate Workflows

# Highway MCP Server - Enable LLMs to Generate Workflows ## Overview Create an MCP SERVER that exposes Highway DSL capabilities to LLMs (Claude, GPT, etc.), enabling them to generate, validate, and submit workflows through natural language. ## Use Case User connects Claude Desktop (or any MCP client) to Highway MCP Server: - "Create a workflow that downloads files from S3, processes them in parallel, and sends an email when done" - Claude uses MCP tools to generate valid Highway DSL Python code - Workflow is validated and optionally submitted ## MCP Server Components ### 1. Tools (Actions LLM Can Take) ```python @mcp.tool() def generate_workflow(description: str, name: str) -> str: """Generate Highway DSL Python code from natural language description.""" # Returns valid Python DSL code @mcp.tool() def validate_workflow(python_code: str) -> dict: """Validate Highway DSL code for syntax and semantic errors.""" # Returns {"valid": bool, "errors": [...], "warnings": [...]} @mcp.tool() def submit_workflow(python_code: str, tenant_id: str = "default") -> dict: """Submit workflow to Highway engine for execution.""" # Returns {"workflow_run_id": "uuid", "status": "pending"} @mcp.tool() def list_templates() -> list[dict]: """List available workflow templates with descriptions.""" @mcp.tool() def get_template(template_name: str) -> str: """Get a specific workflow template as Python code.""" @mcp.tool() def list_available_tools() -> list[dict]: """List all Highway tools (shell, http, docker, email, llm, etc.).""" @mcp.tool() def get_workflow_status(workflow_run_id: str) -> dict: """Check status of a running workflow.""" @mcp.tool() def list_recent_workflows(limit: int = 10) -> list[dict]: """List recent workflow executions.""" ``` ### 2. Resources (Read-Only Data) ```python @mcp.resource("highway://dsl-reference") def get_dsl_reference() -> str: """Complete Highway DSL reference documentation.""" return hwe_dsl_prompt_output # The full DSL prompt @mcp.resource("highway://templates/{name}") def get_template_resource(name: str) -> str: """Get workflow template by name.""" @mcp.resource("highway://tools/{tool_name}") def get_tool_docs(tool_name: str) -> str: """Get documentation for a specific tool.""" @mcp.resource("highway://examples") def get_examples() -> str: """Get workflow examples for common patterns.""" ``` ### 3. Prompts (Reusable Templates) ```python @mcp.prompt() def workflow_generator() -> str: """System prompt for generating Highway workflows.""" return '''You are a Highway Workflow Engine expert. Generate valid Python DSL code using WorkflowBuilder. Always include: from highway_dsl import WorkflowBuilder Always define get_workflow() function returning builder.build() ...''' @mcp.prompt() def workflow_debugger() -> str: """System prompt for debugging workflow issues.""" ``` --- ## Architecture ### Location: `highway_dsl/mcp_server.py` ```python from mcp.server.fastmcp import FastMCP mcp = FastMCP("Highway Workflow Engine") @mcp.tool() def generate_workflow(description: str, name: str) -> str: ... @mcp.resource("highway://dsl-reference") def get_dsl_reference() -> str: ... if __name__ == "__main__": mcp.run(transport="stdio") # For Claude Desktop # or mcp.run(transport="streamable-http", port=8080) # For remote ``` ### CLI Integration ```bash # Run as stdio server (for Claude Desktop) hwe mcp-server # Run as HTTP server hwe mcp-server --http --port 8080 ``` ### Claude Desktop Configuration ```json { "mcpServers": { "highway": { "command": "hwe", "args": ["mcp-server"] } } } ``` --- ## Implementation Details ### Tool: generate_workflow Uses the DSL prompt + examples to guide generation: ```python @mcp.tool() def generate_workflow(description: str, name: str) -> str: """Generate Highway DSL workflow from description. Args: description: Natural language description of the workflow name: Workflow name (snake_case) Returns: Valid Python code using highway_dsl.WorkflowBuilder """ # Load DSL reference dsl_ref = _get_dsl_prompt() # Load relevant examples based on keywords examples = _find_relevant_examples(description) # Return structured prompt for LLM to complete # (The LLM calling this tool will use the output) return f''' Based on this DSL reference: {dsl_ref} And these examples: {examples} Generate a workflow named "{name}" that: {description} ''' ``` ### Tool: validate_workflow ```python @mcp.tool() def validate_workflow(python_code: str) -> dict: """Validate Highway DSL Python code.""" errors = [] warnings = [] # 1. Syntax check try: ast.parse(python_code) except SyntaxError as e: errors.append(f"Syntax error: {e}") return {"valid": False, "errors": errors, "warnings": warnings} # 2. Execute to get workflow object try: namespace = {} exec(python_code, namespace) workflow = namespace.get('get_workflow')() except Exception as e: errors.append(f"Execution error: {e}") return {"valid": False, "errors": errors, "warnings": warnings} # 3. Validate workflow structure validation = workflow.validate() # Built-in validation return { "valid": len(errors) == 0, "errors": errors, "warnings": warnings, "task_count": len(workflow.tasks), "workflow_name": workflow.name } ``` ### Tool: submit_workflow ```python @mcp.tool() def submit_workflow( python_code: str, tenant_id: str = "default", api_url: str = "http://localhost:7822" ) -> dict: """Submit workflow to Highway engine.""" # Validate first validation = validate_workflow(python_code) if not validation["valid"]: return {"error": "Validation failed", "details": validation["errors"]} # Execute to get JSON namespace = {} exec(python_code, namespace) workflow = namespace.get('get_workflow')() workflow_json = workflow.to_json() # Submit via API response = requests.post( f"{api_url}/api/v1/workflows", json={"workflow": json.loads(workflow_json)}, headers={"Authorization": f"Bearer {TOKEN}", "X-Tenant-ID": tenant_id} ) return response.json() ``` --- ## Files to Create - `highway_dsl/mcp_server.py` - Main MCP server implementation - `highway_dsl/mcp_tools.py` - Tool implementations - `highway_dsl/mcp_resources.py` - Resource handlers - `highway_dsl/mcp_prompts.py` - Prompt templates - `tests/test_mcp_server.py` - Tests ## Files to Modify - `highway_dsl/cli.py` - Add `mcp-server` command - `highway_dsl/__init__.py` - Export MCP server - `pyproject.toml` - Add `mcp[cli]` dependency --- ## Example Session User in Claude Desktop: > "Create a workflow that fetches data from 3 APIs in parallel, combines the results, and saves to a file" Claude (using Highway MCP): 1. Calls `list_available_tools()` to see what's available 2. Calls `get_dsl_reference()` resource for syntax 3. Generates Python code using WorkflowBuilder 4. Calls `validate_workflow(code)` to check 5. Returns the code to user (or calls `submit_workflow()` if requested) --- ## Dependencies ``` mcp[cli]>=1.0.0 ``` ## References - MCP Python SDK: https://github.com/modelcontextprotocol/python-sdk - FastMCP docs: https://modelcontextprotocol.io/docs/tools/fastmcp