get_workflow_info
Retrieve detailed information about a specific workflow using its ID on the Unstructured API MCP Server to manage and monitor workflow status and attributes.
Instructions
Get detailed information about a specific workflow.
Args:
workflow_id: ID of the workflow to get information for
Returns:
String containing the workflow information
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| workflow_id | Yes |
Implementation Reference
- uns_mcp/server.py:299-348 (handler)The primary handler implementation for the 'get_workflow_info' tool. Decorated with @mcp.tool() for registration in the FastMCP server. Fetches and formats workflow information from the Unstructured API.@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)