list_workflows
Retrieve and filter workflows from the Unstructured API by destination, source, or status to manage data processing pipelines.
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 workflowsInput Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| destination_id | No | ||
| source_id | No | ||
| status | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
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, )