Unstructured API MCP Server

Instructions

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

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