Unstructured API MCP Server

Instructions

List workflows from the Unstructured API.

Args:
    destination_id: Optional destination connector ID to filter by
    source_id: Optional source connector ID to filter by
    status: Optional workflow status to filter by

Returns:
    String containing the list of workflows

Implementation Reference

• uns_mcp/server.py:251-296 (handler)

  The MCP tool handler for 'list_workflows', decorated with @mcp.tool() for automatic registration. It fetches workflows from the Unstructured API client, applies optional filters for destination_id, source_id, and status, sorts them by name, and returns a formatted list.

  @mcp.tool()
  async def list_workflows(
      ctx: Context,
      destination_id: Optional[str] = None,
      source_id: Optional[str] = None,
      status: Optional[WorkflowState | str] = None,
  ) -> str:
      """
      List workflows from the Unstructured API.
  
      Args:
          destination_id: Optional destination connector ID to filter by
          source_id: Optional source connector ID to filter by
          status: Optional workflow status to filter by
  
      Returns:
          String containing the list of workflows
      """
      client = ctx.request_context.lifespan_context.client
  
      request = ListWorkflowsRequest(destination_id=destination_id, source_id=source_id)
  
      if status:
          try:
              status = WorkflowState(status) if isinstance(status, str) else status
              request.status = status
          except KeyError:
              return f"Invalid workflow status: {status}"
  
      response = await client.workflows.list_workflows_async(request=request)
  
      # Sort workflows by name
      sorted_workflows = sorted(
          response.response_list_workflows,
          key=lambda workflow: workflow.name.lower(),
      )
  
      if not sorted_workflows:
          return "No workflows found"
  
      # Format response
      result = ["Available workflows:"]
      for workflow in sorted_workflows:
          result.append(f"- {workflow.name} (ID: {workflow.id})")
  
      return "\n".join(result)

• uns_mcp/server.py:251-251 (registration)

  The @mcp.tool() decorator registers the list_workflows function as an MCP tool.

  @mcp.tool()

• uns_mcp/server.py:21-35 (schema)

  Imports ListWorkflowsRequest schema used internally by the handler for API requests to list workflows.

  from unstructured_client.models.operations import (
      CancelJobRequest,
      CreateWorkflowRequest,
      DeleteWorkflowRequest,
      GetDestinationRequest,
      GetJobRequest,
      GetSourceRequest,
      GetWorkflowRequest,
      ListDestinationsRequest,
      ListJobsRequest,
      ListSourcesRequest,
      ListWorkflowsRequest,
      RunWorkflowRequest,
      UpdateWorkflowRequest,
  )