Unstructured API MCP Server

Instructions

List jobs via the Unstructured API.

Args:
    workflow_id: Optional workflow ID to filter by
    status: Optional job status to filter by

Returns:
    String containing the list of jobs

Implementation Reference

• uns_mcp/server.py:464-508 (handler)

  The handler function for the 'list_jobs' MCP tool. It is registered via the @mcp.tool() decorator and implements the logic to list jobs from the Unstructured API, optionally filtered by workflow_id and status, sorting by creation time and returning formatted job IDs.

  @mcp.tool()
  async def list_jobs(
      ctx: Context,
      workflow_id: Optional[str] = None,
      status: Optional[JobStatus | str] = None,
  ) -> str:
      """
      List jobs via the Unstructured API.
  
      Args:
          workflow_id: Optional workflow ID to filter by
          status: Optional job status to filter by
  
      Returns:
          String containing the list of jobs
      """
      client = ctx.request_context.lifespan_context.client
  
      request = ListJobsRequest(workflow_id=workflow_id, status=status)
  
      if status:
          try:
              status = JobStatus(status) if isinstance(status, str) else status
              request.status = status
          except KeyError:
              return f"Invalid job status: {status}"
  
      response = await client.jobs.list_jobs_async(request=request)
  
      # Sort jobs by name
      sorted_jobs = sorted(
          response.response_list_jobs,
          key=lambda job: job.created_at,
      )
  
      if not sorted_jobs:
          return "No Jobs found"
  
      # Format response
      result = ["Available Jobs by created time:"]
      for job in sorted_jobs:
          result.append(f"- JOB ID: {job.id}")
  
      return "\n".join(result)

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

  Registration of the list_jobs tool using the @mcp.tool() decorator.

  @mcp.tool()

• uns_mcp/server.py:689-691 (helper)

  Helper usage in gather_workflows_details function extracting list_jobs response.

  jobs: list[JobInformation] = jobs.response_list_jobs
  sources: list[SourceConnectorInformation] = sources.response_list_sources
  destinations: list[DestinationConnectorInformation] = destinations.response_list_destinations

• uns_mcp/server.py:30-30 (schema)

  Import of ListJobsRequest model used for API request schema.

  ListJobsRequest,