update_workflow
Modify an existing workflow configuration by updating its name, schedule, source, destination, or workflow nodes to adapt to changing data processing needs.
Instructions
Update an existing workflow.
Args:
workflow_id: ID of the workflow to update
workflow_config: A Typed Dictionary containing required fields (destination_id,
name, source_id, workflow_type) and non-required fields (schedule, and workflow_nodes)
Returns:
String containing the updated workflow information
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| workflow_id | Yes | ||
| workflow_config | Yes |
Implementation Reference
- uns_mcp/server.py:412-440 (handler)The primary handler for the 'update_workflow' tool. This async function is decorated with @mcp.tool(), which registers it as an MCP tool named 'update_workflow'. It takes workflow_id and workflow_config, uses UnstructuredClient to update the workflow, and returns formatted info or error.@mcp.tool() # WorkflowNode.settings. It can be safely deleted when typing is added. async def update_workflow( ctx: Context, workflow_id: str, workflow_config: CreateWorkflowTypedDict, ) -> str: """Update an existing workflow. Args: workflow_id: ID of the workflow to update workflow_config: A Typed Dictionary containing required fields (destination_id, name, source_id, workflow_type) and non-required fields (schedule, and workflow_nodes) Returns: String containing the updated workflow information """ client = ctx.request_context.lifespan_context.client try: workflow = UpdateWorkflow(**workflow_config) response = await client.workflows.update_workflow_async( request=UpdateWorkflowRequest(workflow_id=workflow_id, update_workflow=workflow), ) info = response.workflow_information return await get_workflow_info(ctx, info.id) except Exception as e: return f"Error updating workflow: {str(e)}"
- uns_mcp/server.py:412-412 (registration)The @mcp.tool() decorator registers the update_workflow function as an MCP tool.@mcp.tool()
- uns_mcp/server.py:300-348 (helper)Helper function called by update_workflow (and others) to retrieve and format detailed workflow information after update.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)
- uns_mcp/server.py:49-49 (schema)Import of CreateWorkflowTypedDict, used as type annotation for workflow_config parameter, defining the input schema structure for the tool.from unstructured_client.models.shared.createworkflow import CreateWorkflowTypedDict