Edit Issue - .issue.db

Title *

Description

# 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

Priority

Status

Due Date (YYYY-MM-DD)

Tags (comma separated)

Related Issues (IDs)

Enter IDs of issues related to this one. They will be linked as 'related'.

Cancel

Edit Issue #241