Prefect MCP Server

Instructions

Implementation Reference

• src/mcp_prefect/flow_run.py:220-261 (handler)

  The main handler function for the 'set_flow_run_state' MCP tool. It maps string state names to Prefect state objects and uses the Prefect client to update the flow run state.

  async def set_flow_run_state(
      flow_run_id: str,
      state: str,
      message: Optional[str] = None,
  ) -> List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]]:
      """
      Set a flow run's state.
      
      Args:
          flow_run_id: The flow run UUID
          state: The new state to set (e.g., "SCHEDULED", "RUNNING", "COMPLETED", "FAILED")
          message: An optional message explaining the state change
          
      Returns:
          Result of the state change operation
      """
      async with get_client() as client:
          state_obj = None
          if state.upper() == "SCHEDULED":
              state_obj = Scheduled(message=message)
          elif state.upper() == "RUNNING":
              state_obj = Running(message=message)
          elif state.upper() == "COMPLETED":
              state_obj = Completed(message=message)
          elif state.upper() == "FAILED":
              state_obj = Failed(message=message)
          elif state.upper() == "PENDING":
              state_obj = Pending(message=message)
          elif state.upper() == "CANCELLED":
              state_obj = Cancelled(message=message)
          else:
              return [types.TextContent(
                  type="text", 
                  text=f"Invalid state '{state}'. Must be one of: SCHEDULED, RUNNING, COMPLETED, FAILED, PENDING, CANCELLED"
              )]
          
          result = await client.set_flow_run_state(
              flow_run_id=UUID(flow_run_id),
              state=state_obj
          )
          
          return [types.TextContent(type="text", text=str(result.dict()))]