Workflows MCP Server

Instructions

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

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,
  }