Unstructured API MCP Server

Instructions

Args:
    workflow_id: ID of the workflow to get information for

Returns:
    String containing the workflow information

Implementation Reference

• uns_mcp/server.py:299-348 (handler)

  The handler function for the 'get_workflow_info' tool. It retrieves workflow details from the Unstructured API client and formats them into a readable string output. Registered via @mcp.tool() decorator.

  @mcp.tool()
  async def get_workflow_info(ctx: Context, workflow_id: str) -> str:
      """Get detailed information about a specific workflow.
  
      Args:
          workflow_id: ID of the workflow to get information for
  
      Returns:
          String containing the workflow information
      """
      client = ctx.request_context.lifespan_context.client
  
      response = await client.workflows.get_workflow_async(
          request=GetWorkflowRequest(workflow_id=workflow_id),
      )
  
      info: WorkflowInformation = response.workflow_information
  
      result = ["Workflow Information:"]
      result.append(f"Name: {info.name}")
      result.append(f"ID: {info.id}")
      result.append(f"Status: {info.status.value}")
      if info.workflow_type is None:
          result.append("Type: Undefined")
      else:
          result.append(f"Type: {info.workflow_type.value}")
  
      result.append("\nSources:")
      for source in info.sources:
          result.append(f"  - {source}")
  
      if info.workflow_type and info.workflow_type == WorkflowType.CUSTOM.value:
          result.append("\nWorkflow Nodes:")
          for node in info.workflow_nodes:
              result.append(f"  - {node.name} (Type: {node.type.value}) (Subtype: {node.subtype}):")
              if node.settings:
                  result.append(f"    Settings: {json.dumps(node.settings, indent=8)}")
  
      result.append("\nDestinations:")
      for destination in info.destinations:
          result.append(f"  - {destination}")
  
      result.append("\nSchedule:")
      if info.schedule.crontab_entries:
          for crontab_entry in info.schedule.crontab_entries:
              result.append(f"  - {crontab_entry.cron_expression}")
      else:
          result.append("  - No crontab entry")
  
      return "\n".join(result)