Unstructured API MCP Server

Instructions

List workflows with finished jobs via the Unstructured API.
Args:
    source_type: Optional source connector type to filter by
    destination_type: Optional destination connector type to filter by
Returns:
    String containing the list of workflows with finished jobs and source and destination
    details

Implementation Reference

• uns_mcp/server.py:561-646 (handler)

  Primary handler function for the tool 'list_workflows_with_finished_jobs'. Decorated with @mcp.tool() for automatic registration in the MCP server. Implements logic to filter and list workflows with completed jobs, optionally by source and destination types.

  @mcp.tool()
  async def list_workflows_with_finished_jobs(
      ctx: Context,
      source_type: Optional[SourceConnectorType | str] = None,
      destination_type: Optional[DestinationConnectorType | str] = None,
  ) -> str:
      """
      List workflows with finished jobs via the Unstructured API.
      Args:
          source_type: Optional source connector type to filter by
          destination_type: Optional destination connector type to filter by
      Returns:
          String containing the list of workflows with finished jobs and source and destination
          details
      """
      if source_type:
          try:
              source_type = (
                  SourceConnectorType(source_type) if isinstance(source_type, str) else source_type
              )
          except KeyError:
              return f"Invalid source type: {source_type}"
      if destination_type:
          try:
              destination_type = (
                  DestinationConnectorType(destination_type)
                  if isinstance(destination_type, str)
                  else destination_type
              )
          except KeyError:
              return f"Invalid destination type: {destination_type}"
  
      client = ctx.request_context.lifespan_context.client
      try:
          workflows_details = await gather_workflows_details(client=client)
      except Exception as e:
          return f"Error retrieving workflows: {str(e)}"
  
      filtered_workflows_details = []
  
      for workflow_details in workflows_details:
          updated_workflow_details = deepcopy(workflow_details)
  
          if source_type:
              updated_workflow_details.sources = [
                  source for source in workflow_details.sources if source.type == source_type
              ]
  
          if destination_type:
              updated_workflow_details.destinations = [
                  destination
                  for destination in workflow_details.destinations
                  if destination.type == destination_type
              ]
  
          updated_workflow_details.jobs = [
              job for job in workflow_details.jobs if job.status == JobStatus.COMPLETED
          ]
  
          if (
              updated_workflow_details.sources
              and updated_workflow_details.destinations
              and updated_workflow_details.jobs
          ):
              filtered_workflows_details.append(updated_workflow_details)
  
      if not filtered_workflows_details:
          return "No workflows found with finished jobs"
  
      result = ["Workflows:"]
      for workflow_details in filtered_workflows_details:
          result.append(f"- Name: {workflow_details.workflow.name}")
          result.append(f"  ID: {workflow_details.workflow.id}")
          result.append("  Sources:")
          for source in workflow_details.sources:
              result.append(f"    - {source.name} (ID: {source.id})")
              for key, value in source.config:
                  result.append(f"      {key}: {value}")
  
          result.append("  Destinations:")
          for destination in workflow_details.destinations:
              result.append(f"    - {destination.name} (ID: {destination.id})")
              for key, value in destination.config:
                  result.append(f"      {key}: {value}")
  
      return "\n".join(result)

• uns_mcp/server.py:681-723 (helper)

  Key helper function that aggregates workflow details by fetching lists of workflows, jobs, sources, and destinations from the Unstructured API client and associating them appropriately.

  async def gather_workflows_details(client: UnstructuredClient) -> list[WorkflowDetails]:
      workflows, jobs, sources, destinations = await asyncio.gather(
          client.workflows.list_workflows_async(request=ListWorkflowsRequest()),
          client.jobs.list_jobs_async(request=ListJobsRequest()),
          client.sources.list_sources_async(request=ListSourcesRequest()),
          client.destinations.list_destinations_async(request=ListDestinationsRequest()),
      )
      workflows: list[WorkflowInformation] = workflows.response_list_workflows
      jobs: list[JobInformation] = jobs.response_list_jobs
      sources: list[SourceConnectorInformation] = sources.response_list_sources
      destinations: list[DestinationConnectorInformation] = destinations.response_list_destinations
  
      workflow_id_to_jobs = {
          workflow_id: list(grouped_jobs)
          for workflow_id, grouped_jobs in groupby(jobs, lambda x: x.workflow_id)
      }
      source_id_to_source_info = {source.id: source for source in sources}
      destination_id_to_destination_info = {
          destination.id: destination for destination in destinations
      }
  
      sorted_workflows = sorted(workflows, key=lambda x: x.updated_at, reverse=True)
  
      workflows_details = []
  
      for workflow in sorted_workflows:
          workflow_details = WorkflowDetails(
              workflow=workflow,
              jobs=list(workflow_id_to_jobs.get(workflow.id, [])),
              sources=[
                  source_id_to_source_info[source_id]
                  for source_id in workflow.sources
                  if source_id in source_id_to_source_info
              ],
              destinations=[
                  destination_id_to_destination_info[destination_id]
                  for destination_id in workflow.destinations
                  if destination_id in destination_id_to_destination_info
              ],
          )
          workflows_details.append(workflow_details)
  
      return workflows_details

• uns_mcp/server.py:674-678 (schema)

  Pydantic model defining the structure for workflow details used in the tool implementation.

  class WorkflowDetails(BaseModel):
      workflow: WorkflowInformation
      jobs: list[JobInformation]
      sources: list[SourceConnectorInformation]
      destinations: list[DestinationConnectorInformation]

• uns_mcp/server.py:726-750 (registration)

  Entry point where the MCP server is run, automatically registering all @mcp.tool() decorated functions including the target tool.

  def main():
      load_environment_variables()
      if len(sys.argv) < 2:
          # server is directly being invoked from client
          mcp.run(transport="stdio")
      else:
          # server is running as HTTP SSE server
          # reference: https://github.com/sidharthrajaram/mcp-sse
          mcp_server = mcp._mcp_server  # noqa: WPS437
  
          import argparse
  
          parser = argparse.ArgumentParser(description="Run MCP SSE-based server")
          parser.add_argument("--host", default="127.0.0.1", help="Host to bind to")
          parser.add_argument("--port", type=int, default=8080, help="Port to listen on")
          args = parser.parse_args()
  
          # Bind SSE request handling to MCP server
          starlette_app = create_starlette_app(mcp_server, debug=True)
  
          uvicorn.run(starlette_app, host=args.host, port=args.port)
  
  
  if __name__ == "__main__":
      main()