Workflows MCP Server

Instructions

Implementation Reference

• src/server.py:229-261 (handler)

  Primary MCP tool handler for reading a workflow's source code. Uses get_workflow_path helper to locate the file and returns its content.
  @mcp.tool() def read_workflow(name: str) -> dict: """ Read the source code of a workflow script. Args: name: The name of the workflow to read Returns: dict: The workflow source code and metadata """ try: workflow_path = get_workflow_path(name) if not workflow_path.exists(): return { "status": "error", "message": f"Workflow '{name}' not found" } content = workflow_path.read_text() return { "status": "success", "name": name, "path": str(workflow_path), "content": content } except Exception as e: return { "status": "error", "message": f"Failed to read workflow: {str(e)}" }
• src/http_server.py:293-315 (handler)

  HTTP handler for read_workflow tool in REST API server.
  async def read_workflow_handler(body: dict): """Handle read_workflow tool call.""" try: name = body.get("name") if not name: return {"status": "error", "message": "Workflow name is required"} workflow_path = get_workflow_path(name) if not workflow_path.exists(): return {"status": "error", "message": f"Workflow '{name}' not found"} content = workflow_path.read_text() return { "status": "success", "name": name, "path": str(workflow_path), "content": content } except Exception as e: return {"status": "error", "message": str(e)}
• src/mcp_http_server.py:291-309 (handler)

  Tool handler for read_workflow in MCP HTTP SSE server.
  def tool_read_workflow(arguments: dict) -> dict: """Read a workflow's source code.""" try: name = arguments.get("name") if not name: return {"error": "Workflow name is required"} workflow_path = get_workflow_path(name) if not workflow_path.exists(): return {"error": f"Workflow '{name}' not found"} content = workflow_path.read_text() return {"name": name, "content": content} except Exception as e: return {"error": str(e)}
• src/mcp_http_server.py:127-147 (schema)

  Input schema definition for the read_workflow tool in MCP_TOOLS list.
  "name": "list_workflows", "description": "List all available workflow scripts with their metadata.", "inputSchema": { "type": "object", "properties": {} } }, { "name": "read_workflow", "description": "Read the source code of a workflow script.", "inputSchema": { "type": "object", "properties": { "name": { "type": "string", "description": "The name of the workflow to read" } }, "required": ["name"] } },
• src/mcp_http_server.py:378-385 (registration)

  Registration of read_workflow handler in TOOL_HANDLERS dictionary.
  TOOL_HANDLERS = { "create_workflow": tool_create_workflow, "execute_workflow": tool_execute_workflow, "list_workflows": tool_list_workflows, "read_workflow": tool_read_workflow, "update_workflow": tool_update_workflow, "delete_workflow": tool_delete_workflow, }